You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/12 15:02:45 UTC
(iggy) branch master updated: feat(gateways): Kafka Gateway add DescribeGroups and ListGroups Keys (#4401)
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 7a08127a8 feat(gateways): Kafka Gateway add DescribeGroups and ListGroups Keys (#4401)
7a08127a8 is described below
commit 7a08127a8b7f11e764d735b0c18dfb0a76dd02d0
Author: ryerraguntla <ry...@users.noreply.github.com>
AuthorDate: Mon Oct 12 11:02:40 2026 -0400
feat(gateways): Kafka Gateway add DescribeGroups and ListGroups Keys (#4401)
---
gateways/kafka/README.md | 7 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 45 +-
gateways/kafka/docs/MANUAL_TESTING.md | 13 +-
gateways/kafka/docs/SCOPE.md | 12 +-
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/docs/kafka_api_keys_reference.md | 15 +-
gateways/kafka/scripts/ci-wire-fixtures.sh | 2 +-
gateways/kafka/src/group/mod.rs | 85 +-
gateways/kafka/src/group/state.rs | 853 ++++++++++++++++++++-
gateways/kafka/src/protocol/api.rs | 15 +-
gateways/kafka/src/protocol/bounds_guard.rs | 125 ++-
gateways/kafka/src/protocol/dedup.rs | 73 ++
.../kafka/src/protocol/handlers/describe_groups.rs | 165 ++++
.../kafka/src/protocol/handlers/list_groups.rs | 246 ++++++
gateways/kafka/src/protocol/handlers/metadata.rs | 16 +-
gateways/kafka/src/protocol/handlers/mod.rs | 15 +-
gateways/kafka/src/protocol/mod.rs | 2 +
gateways/kafka/src/protocol/wire_len.rs | 83 ++
gateways/kafka/tests/common/scope.rs | 2 +
gateways/kafka/tests/common/wire.rs | 50 +-
gateways/kafka/tests/consumer_group_tests.rs | 675 +++++++++++++++-
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 16 +-
gateways/kafka/tests/server_e2e_tests.rs | 4 +-
gateways/kafka/tests/version_firewall_tests.rs | 30 +-
24 files changed, 2461 insertions(+), 90 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 6c729ce7d..3b88850b7 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -20,8 +20,11 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> 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.
>
> Consumer group coordination is not a stub either: `FindCoordinator`, `JoinGroup`, `Heartbeat`,
-> `LeaveGroup` and `SyncGroup` are real, with real membership, rebalances, graceful leave and
-> session expiry ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). With the bridge off,
+> `LeaveGroup`, `SyncGroup`, `DescribeGroups` and `ListGroups` are real, with real membership,
+> rebalances, graceful leave, session expiry and admin views of that membership
+> ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). `kafka-consumer-groups.sh --list`,
+> `--describe --members` and `--describe --state` use those views. The default `--describe`
+> calls OffsetFetch (API key 9), which is not implemented, so that mode does not work. With the bridge off,
> Metadata reports every topic unknown, so a consumer joins a group and is assigned 0 partitions.
> With the bridge on, OffsetCommit and OffsetFetch keep each group's offsets in Iggy, so a group
> resumes from its last commit, also after a gateway restart
diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md
index a09f17919..166f6acac 100644
--- a/gateways/kafka/docs/CONSUMER_GROUPS.md
+++ b/gateways/kafka/docs/CONSUMER_GROUPS.md
@@ -2,7 +2,9 @@
What [#3541](https://github.com/apache/iggy/issues/3541) added: `FindCoordinator` (10),
`JoinGroup` (11), `Heartbeat` (12) and `SyncGroup` (14), backed by an in-memory coordinator in
-`src/group/`. [#3543](https://github.com/apache/iggy/issues/3543) added `LeaveGroup` (13). This is Kafka's *classic* group protocol. Offsets are a separate concern and live
+`src/group/`. [#3543](https://github.com/apache/iggy/issues/3543) added `LeaveGroup` (13).
+[#3548](https://github.com/apache/iggy/issues/3548) added `DescribeGroups` (15) and `ListGroups`
+(16). This is Kafka's *classic* group protocol. Offsets are a separate concern and live
in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)).
| API key | Name | Versions | Notes |
@@ -12,11 +14,48 @@ in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)).
| 12 | Heartbeat | 0-4 | Refreshes a session; `REBALANCE_IN_PROGRESS` is how a follower learns to rejoin |
| 13 | LeaveGroup | 0-5 | Removes members; survivors' parked joins and syncs are released at once |
| 14 | SyncGroup | 0-5 | Relays the leader's assignment blobs; a follower parks until the leader syncs |
+| 15 | DescribeGroups | 0-6 | Members, assignment and state. A group that is not here is `Dead`; v6 also returns `GROUP_ID_NOT_FOUND` (69) |
+| 16 | ListGroups | 0-5 | Groups currently on this coordinator. v4+ can filter by state, v5 by type `classic` |
`kafka-protocol` can encode FindCoordinator v5 and v6 as well, and they are byte-identical to v4.
They are not advertised because `SCOPE.md`'s governance model only admits a version once it has
been manually tested.
+## Admin views
+
+`DescribeGroups` and `ListGroups` read the same in-memory map the join path writes. State strings
+are `Empty`, `PreparingRebalance`, `CompletingRebalance` and `Stable`. `Empty` means the group
+has no joined members: a v4+ client has been handed a member id and has not rejoined with it yet.
+`Dead` is only the describe answer for an id that is not in the map.
+
+A group with neither members nor pending member ids is removed. Kafka keeps an `Empty` group
+that only has committed offsets. Offsets are not stored here yet.
+TODO(#3542): once offset storage lands, keep those groups and report `Empty`.
+
+`ListGroups` does not tick. It leaves sessions, pending ids and phases untouched, and it omits a
+group whose member sessions and pending-id deadlines are all already past. `DescribeGroups` does
+tick each named group, so that same id is then `Dead`.
+
+While a group is not `Stable`, `DescribeGroups` leaves `protocol_data`, member metadata and
+member assignment empty. Kafka does the same: during `PreparingRebalance` the stored assignment
+is the previous generation, and a rejoined member has already replaced its subscription. A
+`Stable` group whose own description does not fit `max_frame_size` is returned in that same
+shape. Further groups that do not fit the remaining response are returned with a per-group
+`UNKNOWN_SERVER_ERROR` (-1). The connection stays open.
+
+A group id repeated in one `DescribeGroups` request is answered once, not once per repeat: a
+second snapshot of the same members would otherwise multiply the encoded response by the repeat
+count.
+
+`kafka-consumer-groups.sh --list`, `--describe --members` and `--describe --state` use these
+two APIs. The default `--describe` calls `OffsetFetch` (API key 9), which is not implemented,
+so that mode does not work. `--reset-offsets` needs the same API.
+
+Member `client_id` and `client_host` are empty. The coordinator does not keep the request
+header's client id or the connection's peer address. `include_authorized_operations` is answered
+with the omitted sentinel. There is no group ACL bitfield. `ListGroups` v5 sets `group_type` to
+`classic`, which is the only protocol this coordinator runs.
+
## Assignment is the client's job
The gateway elects a leader, hands it every member's subscription metadata, and fans the blobs the
@@ -82,7 +121,9 @@ survivors keep the current generation and learn about the rebalance through
rejoin rather than as lost partitions.
When a group empties, by leave or by expiry, it is dropped, so the next group under that name
-starts again at generation 1. Kafka instead keeps the `Empty` group and its generation. This is
+starts again at generation 1. The pending-member-id window above is the one case that stays and
+reports `Empty`. Kafka also keeps an `Empty` group that only has committed offsets, and its
+generation with it. This coordinator does not, until offset storage lands (#3542). This is
safe because member ids carry a UUID and every generation check is preceded by a membership
check, so a stale client is told `UNKNOWN_MEMBER_ID` before its generation is ever compared.
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 072eea17a..36a6602cc 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -224,6 +224,8 @@ For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-mes
| 12 | Heartbeat | 0 | 4 | −1, 0, 4, 5 |
| 13 | LeaveGroup | 0 | 5 | −1, 0, 5, 6 |
| 14 | SyncGroup | 0 | 5 | −1, 0, 5, 6 |
+| 15 | DescribeGroups | 0 | 6 | −1, 0, 6, 7 |
+| 16 | ListGroups | 0 | 5 | −1, 0, 5, 6 |
| 22 | InitProducerId | 0 | 5 | −1, 0, 5, 6 |
| 32 | DescribeConfigs | 1 | 4 | 0, 1, 4, 5 |
| 33 | AlterConfigs | 0 | 2 | −1, 0, 2, 3 |
@@ -232,8 +234,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/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) |
+| 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 eight 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,15,16,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
@@ -324,6 +326,9 @@ success. This category predates the bridge landing: G2/G3's "fails at metadata"
| G9 | Group resumes from its commit | Bridge on and topic `orders` made as in Category I. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g9 --topic orders --from-beginning`, stop it, produce one record, run it again | The second run prints only the new record. `--from-beginning` applies only to a group with no committed offset |
| G10 | Resume after a gateway restart | G9 with a gateway restart before the second run | Same as G9, because the offsets live in Iggy ([#3542](https://github.com/apache/iggy/issues/3542)) |
| G11 | Two members share a topic | Bridge on and a topic with 2 partitions. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g11 --topic <topic> --property print.partition=true --property print.offset=true` in two terminals, produce to both partitions, stop one, produce again | Each prints one partition. After the stop, the other prints the new records of both partitions, and none twice |
+| G12 | List groups | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list` after G5 has joined | Prints the group. `--list --state Empty` matches a group that was issued a member id and has not joined yet. A group whose members have all expired is absent |
+| G13 | Describe members | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --members --group g2` | Prints member ids. Assignment and protocol are filled only while the group is Stable |
+| G14 | Describe state | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --state --group g2` | Prints the state string (`Stable`, `PreparingRebalance`, `CompletingRebalance`, or `Empty`). The default `--describe` (no `--members` or `--state`) calls OffsetFetch and fails, because that API is not implemented |
Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke.
@@ -453,7 +458,7 @@ kcat version (if used): ___________
[ ] D1–D10 Flexible vs legacy encoding
[ ] E1–E4 Metadata stub semantics
[ ] F1–F6 TCP / connection behavior
-[ ] G1–G10 kcat / Java client (record errors for G2/G3)
+[ ] G1–G11 kcat / Java client / kafka-consumer-groups.sh (record errors for G2/G3; G11's default `--describe` is expected to fail)
[ ] H1–H6 Adversarial input
[ ] I1–I6 Docker Compose quick start (record wall-clock time to I3; I5 is restart durability)
@@ -489,4 +494,6 @@ These are documented as TODO in [SCOPE.md](SCOPE.md) — do not fail #3421 valid
- SASL authentication
- Accurate partition leadership / ISR
- Transactional produce
+- Real offset commit semantics
+- Consumer group offset commit/fetch, including `kafka-consumer-groups.sh --describe` without `--members` or `--state` (list, `--describe --members` and `--describe --state` are G9-G11; join/sync/heartbeat/leave are G3-G8)
- Admin group views (join/sync/heartbeat/leave are covered by G3-G8, offset commit and fetch by G9-G10)
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 0f1695466..91d2e7e36 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -35,7 +35,7 @@ Expand `SUPPORTED_RANGES` only after a key/version pair is manually tested. ApiV
**Every unsupported-version case closes the connection, for every listed key** - not just above
the encoder max. `kafka_protocol`'s schema floor for each supported message happens to equal
`SUPPORTED_RANGES`' own min today (Produce 3, Fetch 4, ListOffsets 1, Metadata 0, ApiVersions 0,
-CreateTopics 2, InitProducerId 0, OffsetCommit 2, OffsetFetch 1,DescribeConfigs 1, AlterConfigs 0, and 0 for the five coordination keys), so there is no version below an API's
+CreateTopics 2, InitProducerId 0, OffsetCommit 2, OffsetFetch 1, DescribeConfigs 1, AlterConfigs 0, and 0 for the five group coordination keys, DescribeGroups and ListGroups), so there is no version below an API's
min that the crate can actually encode a response for either - `unsupported_version_response`
still tries, but the encode attempt fails and the connection closes rather than sending a
malformed body.
@@ -63,6 +63,8 @@ it knows the server supports flexible encoding.
| 12 | Heartbeat | 0 | 4 | 0, 1, 2, 3, 4 | Refreshes a session; `REBALANCE_IN_PROGRESS` (27) drives a rejoin; flexible encoding at v4+ |
| 13 | LeaveGroup | 0 | 5 | 0, 1, 2, 3, 4, 5 | Removes members, per-member errors from v3; flexible encoding at v4+ |
| 14 | SyncGroup | 0 | 5 | 0, 1, 2, 3, 4, 5 | Relays the leader's assignment blobs; flexible encoding at v4+ |
+| 15 | DescribeGroups | 0 | 6 | 0, 1, 2, 3, 4, 5, 6 | Members, assignment and state of groups on this coordinator; a missing group is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69); flexible encoding at v5+ |
+| 16 | ListGroups | 0 | 5 | 0, 1, 2, 3, 4, 5 | Groups on this coordinator; v4+ state filter and v5 type filter; flexible encoding at v3+ |
| 22 | InitProducerId | 0 | 5 | 0, 1, 2, 3, 4, 5 | Allocate a producer id (epoch 0); a `transactional_id` gets `UNSUPPORTED_VERSION` (35); flexible encoding at v2+ |
| 32 | DescribeConfigs | 1 | 4 | 1, 2, 3, 4 | Topic resources only. Returns `retention.ms` and `cleanup.policy`. Flexible encoding at v4. See [`CONFIGS.md`](CONFIGS.md) |
| 33 | AlterConfigs | 0 | 2 | 0, 1, 2 | Persists `retention.ms` only. Flexible encoding at v2. See [`CONFIGS.md`](CONFIGS.md) |
@@ -88,6 +90,8 @@ Use this table when configuring clients or generating wire fixtures with `kafka-
| 12 | Heartbeat | 0–4 | v4 |
| 13 | LeaveGroup | 0–5 | v4 |
| 14 | SyncGroup | 0–5 | v4 |
+| 15 | DescribeGroups | 0–6 | v5 |
+| 16 | ListGroups | 0–5 | v3 |
| 18 | ApiVersions | 0–3 | v3 |
| 19 | CreateTopics | 2–5 | v5 |
| 20 | DeleteTopics | 1–5 | v4 |
@@ -103,7 +107,6 @@ All API keys not listed above close the connection (see Governance model above)
| API key | Name | Notes |
| --------- | ------ | ------- |
-| 15, 16 | DescribeGroups, ListGroups | Admin views — [#3548](https://github.com/apache/iggy/issues/3548) |
| 17 | SaslHandshake | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`AUTHENTICATION.md`](AUTHENTICATION.md)) |
| 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)) |
@@ -171,7 +174,7 @@ 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 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 |
+| **2 — Request/response codecs** | Partial | Decode/encode via the `kafka_protocol` crate (broker feature only) for 19 keys; `bounds_guard.rs` pre-validates against unbounded allocation before handing a frame to the crate; stub responses except InitProducerId and the 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 [...]
---
@@ -393,9 +396,10 @@ Offset persistence design ([#3540](https://github.com/apache/iggy/issues/3540)):
- [x] FindCoordinator (10), JoinGroup (11), Heartbeat (12), SyncGroup (14) -
[#3541](https://github.com/apache/iggy/issues/3541); LeaveGroup (13) -
[#3543](https://github.com/apache/iggy/issues/3543); see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)
+- [x] DescribeGroups (15), ListGroups (16) -
+ [#3548](https://github.com/apache/iggy/issues/3548); see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)
- [x] OffsetCommit (8), OffsetFetch (9) - [#3542](https://github.com/apache/iggy/issues/3542);
offsets live in Iggy, see [`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)
-- [ ] DescribeGroups (15), ListGroups (16) as needed by target clients
### Phase 3+ — Auth, admin, tuning
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index 8ce0421dc..0a62468fd 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -74,7 +74,7 @@ file under `tests/` anymore.
| [`idempotence_tests.rs`](../tests/idempotence_tests.rs) | `InitProducerId` allocation across every supported version, and the transactional refusals on `InitProducerId`/Produce | No |
| [`broker_advertise_tests.rs`](../tests/broker_advertise_tests.rs) | `BrokerAdvertise::from_server_config` parsing | No |
| [`server_integration_tests.rs`](../tests/server_integration_tests.rs) | `read_frame` unit-level I/O | No |
-| [`consumer_group_tests.rs`](../tests/consumer_group_tests.rs) | `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup` against one shared `GatewayState` with paused time, plus two-socket TCP rebalance and leave scenarios | No |
+| [`consumer_group_tests.rs`](../tests/consumer_group_tests.rs) | `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup`/`DescribeGroups`/`ListGroups` against one shared `GatewayState` with paused time, plus two-socket TCP rebalance and leave scenarios | No |
| [`server_e2e_tests.rs`](../tests/server_e2e_tests.rs) | Full `KafkaGateway` TCP round-trips | Partial |
| [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No |
| [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md
index 4fc957ed3..59f4f2f9d 100644
--- a/gateways/kafka/docs/kafka_api_keys_reference.md
+++ b/gateways/kafka/docs/kafka_api_keys_reference.md
@@ -88,10 +88,13 @@ Key new minimums:
| 12 | **Heartbeat** | 1 | 4 | v4 | 🟠 Required Stub |
| 13 | **LeaveGroup** | 1 | 5 | v4 | 🟠 Required Stub |
| 14 | **SyncGroup** | 1 | 5 | v4 | 🟠 Required Stub |
-| 15 | **DescribeGroups** | 0 | 6 | v5 | 🟡 Optional Stub |
-| 16 | **ListGroups** | 1 | 5 | v3 | 🟡 Optional Stub |
+| 15 | **DescribeGroups** | 0 | 6 | v5 | Implemented |
+| 16 | **ListGroups** | 1 | 5 | v3 | Implemented |
| 42 | **DeleteGroups** | 1 | 2 | v2 | 🟡 Optional Stub |
+> DescribeGroups and ListGroups are implemented. They read this coordinator's in-memory groups.
+> See [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md).
+
---
## Group 4 — New Consumer Group Protocol (KIP-848, Kafka 3.7+)
@@ -268,8 +271,9 @@ Key new minimums:
| Category | Count | Notes |
| ---------- | :-----: | ------- |
| 🔴 Bridge (data path) | 9 | Produce, Fetch, Metadata, OffsetCommit, OffsetFetch, SaslHandshake, ApiVersions, SaslAuthenticate, ShareFetch |
-| 🟠 Required Stub (client state machine) | 11 | ListOffsets, consumer group (10-14), CreateTopics, InitProducerId (22), ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) |
-| 🟡 Optional Stub (admin/observability) | 41 | Can return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` safely |
+| 🟠 Required Stub (client state machine) | 13 | ListOffsets, consumer group (8-14), CreateTopics, InitProducerId (22), ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) |
+| Implemented | 2 | DescribeGroups (15), ListGroups (16) |
+| 🟡 Optional Stub (admin/observability) | 39 | Can return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` safely |
| ❌ Unadvertised | 5 | AddPartitionsToTxn (24), AddOffsetsToTxn (25), EndTxn (26), TxnOffsetCommit (28), and DeleteRecords (21). Absent from ApiVersions, so a conforming client never sends one |
| ❌ Reject (broker/KRaft internal) | 22 | Return `INVALID_REQUEST` with valid frame — never close the TCP connection |
| **Total API Keys in this document** | **88** | Key IDs 0-88 with a gap at 73 |
@@ -309,7 +313,8 @@ be agreed. This includes:
- **All broker/KRaft-internal keys** (Group 14)
The classic consumer group keys FindCoordinator (10), JoinGroup (11), Heartbeat (12),
-LeaveGroup (13) and SyncGroup (14) are supported - see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md).
+LeaveGroup (13), SyncGroup (14), DescribeGroups (15) and ListGroups (16) are supported - see
+[`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md).
OffsetCommit (8) and OffsetFetch (9) are supported too - see [`OFFSET_STORAGE.md`](OFFSET_STORAGE.md).
Remaining scope (consumer groups, auth, admin/tuning) is tracked in `SCOPE.md`'s TODO section,
diff --git a/gateways/kafka/scripts/ci-wire-fixtures.sh b/gateways/kafka/scripts/ci-wire-fixtures.sh
index 5251e8392..636902224 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 20 22 32 33)
+FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 15 16 19 20 22 32 33)
usage() {
echo "Usage: $0 {generate|cleanup}" >&2
diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index b59cc7787..4af5d481a 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -19,8 +19,8 @@
//!
//! [`GroupCoordinator`] owns every group this gateway instance coordinates and is the only
//! module that awaits: `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup`/
-//! `OffsetCommit` handlers translate wire messages into the request types here, and `state` holds
-//! the synchronous state machine those requests drive.
+//! `DescribeGroups`/`ListGroups`/`OffsetCommit` handlers translate wire messages into the request types here,
+//! and `state` holds the synchronous state machine those requests drive.
//!
//! Membership is process memory, not Iggy state. Two gateway instances fronting one Iggy cluster
//! therefore coordinate two independent groups under one name; see `docs/CONSUMER_GROUPS.md`.
@@ -44,6 +44,22 @@ use tokio_util::sync::CancellationToken;
use crate::group::state::{GroupState, Step};
use crate::protocol::api::{ERROR_NONE, ERROR_NOT_COORDINATOR, ERROR_UNKNOWN_MEMBER_ID};
+/// `DescribeGroups` / `ListGroups` state string for a group with no joined members.
+pub const GROUP_STATE_EMPTY: &str = "Empty";
+/// `DescribeGroups` / `ListGroups` state string for a group preparing a rebalance.
+pub const GROUP_STATE_PREPARING_REBALANCE: &str = "PreparingRebalance";
+/// `DescribeGroups` / `ListGroups` state string for a group waiting on `SyncGroup`.
+pub const GROUP_STATE_COMPLETING_REBALANCE: &str = "CompletingRebalance";
+/// `DescribeGroups` / `ListGroups` state string for a group whose assignment is in effect.
+pub const GROUP_STATE_STABLE: &str = "Stable";
+
+/// v6 `error_message` when one `DescribeGroups` group does not fit the response budget.
+///
+/// `UNKNOWN_SERVER_ERROR` is not retried by `kafka-consumer-groups.sh`, so the tool prints this
+/// for the groups that did not fit and still shows the ones that did. A retriable code would
+/// send the same describe-all request back until the client timed out.
+pub const DESCRIBE_RESPONSE_TOO_LARGE: &str = "Group description exceeds the response size limit.";
+
/// Kafka's own `group.min.session.timeout.ms` default.
const DEFAULT_MIN_SESSION_TIMEOUT: Duration = Duration::from_secs(6);
/// Kafka's own `group.max.session.timeout.ms` default.
@@ -324,6 +340,40 @@ impl LeaveResult {
}
}
+/// One member as `DescribeGroups` reports it.
+#[derive(Debug, Clone)]
+pub struct MemberDescription {
+ pub member_id: StrBytes,
+ pub group_instance_id: Option<StrBytes>,
+ /// Subscription metadata for the group's selected protocol. Empty until a protocol is chosen.
+ pub metadata: Bytes,
+ pub assignment: Bytes,
+}
+
+/// One group as `ListGroups` reports it. Member blobs stay on the coordinator.
+#[derive(Debug, Clone)]
+pub struct GroupListing {
+ pub group_id: StrBytes,
+ pub protocol_type: StrBytes,
+ pub state: &'static str,
+}
+
+/// One group the coordinator currently holds, as `DescribeGroups` reports it.
+#[derive(Debug, Clone)]
+pub struct GroupDescription {
+ pub group_id: StrBytes,
+ /// `ERROR_NONE`, or `ERROR_UNKNOWN_SERVER_ERROR` when this group does not fit the response.
+ /// A non-zero code is encoded as an empty group: the other fields are not put on the wire.
+ pub error: i16,
+ /// One of the `GROUP_STATE_*` strings.
+ pub state: &'static str,
+ pub protocol_type: StrBytes,
+ /// Selected protocol name. `None` until the group is `Stable`, and `None` when the group is
+ /// described without member metadata.
+ pub protocol_name: Option<StrBytes>,
+ pub members: Vec<MemberDescription>,
+}
+
/// The answer for one `LeavingMember`.
#[derive(Debug, Clone)]
pub struct LeftMember {
@@ -492,6 +542,28 @@ impl GroupCoordinator {
state::leave_step(&mut groups, request, Instant::now())
}
+ /// Snapshot of each distinct group, in first-seen order. `None` means the group is not here.
+ ///
+ /// Ids are deduped before the lock is taken. Under the lock each remaining group is ticked
+ /// once and snapshotted once. A group that does not fit is returned with
+ /// `UNKNOWN_SERVER_ERROR` instead of failing the whole response. Encoding happens
+ /// after this returns.
+ pub async fn describe_groups(
+ &self,
+ group_ids: &[StrBytes],
+ version: i16,
+ max_frame_size: usize,
+ ) -> Vec<Option<GroupDescription>> {
+ let mut groups = self.groups.lock().await;
+ state::describe_groups(
+ &mut groups,
+ group_ids,
+ version,
+ max_frame_size,
+ Instant::now(),
+ )
+ }
+
/// Whether `request` may commit offsets for its group. A member's commit also refreshes its
/// session. Never parks.
pub async fn validate_commit(&self, request: &CommitRequest) -> i16 {
@@ -510,6 +582,15 @@ impl GroupCoordinator {
let groups = self.groups.lock().await;
state::offset_hold(&groups, member, hold, Instant::now())
}
+ /// Every group currently in the map, ordered by group id.
+ ///
+ /// Does not tick. A list names no group, and sweeping here would open rebalances as a side
+ /// effect of reading. Groups whose sessions and pending ids are already past `now` are left
+ /// out, which is the same membership `DescribeGroups` would see after it did tick.
+ pub async fn list_groups(&self) -> Vec<GroupListing> {
+ let groups = self.groups.lock().await;
+ state::list_groups(&groups, Instant::now())
+ }
/// Sleeps until the group changes or `wake_at` passes. `false` means the gateway is draining.
async fn wait_until(&self, mut receiver: watch::Receiver<u64>, wake_at: Instant) -> bool {
diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs
index 92d057f95..9106e6daa 100644
--- a/gateways/kafka/src/group/state.rs
+++ b/gateways/kafka/src/group/state.rs
@@ -22,8 +22,9 @@
//! the protocol rules are testable without a runtime and the coordinator's lock is never held
//! across an `.await`.
//!
-//! Kafka's `Empty` group state is "absent from the map": offsets live in Iggy, so an empty group
-//! holds nothing worth keeping and retaining it would be an unbounded-memory vector.
+//! A group with no joined members is reported as Kafka's `Empty`. One with neither members nor
+//! pending member ids is dropped: offsets live in Iggy, so nothing here is worth keeping.
+//! TODO(#3542): keep an offsets-only group and report `Empty` once those offsets are stored here.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::time::Duration;
@@ -35,15 +36,22 @@ use tokio::time::Instant;
use uuid::Uuid;
use crate::group::{
- CommitRequest, GroupCoordinatorConfig, GroupMember, JoinRequest, JoinResult, JoinedMember,
- LeaveRequest, LeaveResult, LeavingMember, LeftMember, SyncRequest, SyncResult, owned_str,
+ CommitRequest, DESCRIBE_RESPONSE_TOO_LARGE, GROUP_STATE_COMPLETING_REBALANCE,
+ GROUP_STATE_EMPTY, GROUP_STATE_PREPARING_REBALANCE, GROUP_STATE_STABLE, GroupCoordinatorConfig,
+ GroupDescription, GroupListing, GroupMember, JoinRequest, JoinResult, JoinedMember,
+ LeaveRequest, LeaveResult, LeavingMember, LeftMember, MemberDescription, SyncRequest,
+ SyncResult, owned_str,
};
use crate::protocol::api::{
- ERROR_COORDINATOR_NOT_AVAILABLE, ERROR_FENCED_INSTANCE_ID, ERROR_GROUP_MAX_SIZE_REACHED,
- ERROR_ILLEGAL_GENERATION, ERROR_INCONSISTENT_GROUP_PROTOCOL, ERROR_INVALID_GROUP_ID,
- ERROR_INVALID_REQUEST, ERROR_INVALID_SESSION_TIMEOUT, ERROR_MEMBER_ID_REQUIRED, ERROR_NONE,
- ERROR_REBALANCE_IN_PROGRESS, ERROR_UNKNOWN_MEMBER_ID,
+ API_KEY_DESCRIBE_GROUPS, ERROR_COORDINATOR_NOT_AVAILABLE, ERROR_FENCED_INSTANCE_ID,
+ ERROR_GROUP_MAX_SIZE_REACHED, ERROR_ILLEGAL_GENERATION, ERROR_INCONSISTENT_GROUP_PROTOCOL,
+ ERROR_INVALID_GROUP_ID, ERROR_INVALID_REQUEST, ERROR_INVALID_SESSION_TIMEOUT,
+ ERROR_MEMBER_ID_REQUIRED, ERROR_NONE, ERROR_REBALANCE_IN_PROGRESS, ERROR_UNKNOWN_MEMBER_ID,
+ ERROR_UNKNOWN_SERVER_ERROR,
};
+use crate::protocol::dedup::dedup_first_seen;
+use crate::protocol::header::response_header_version;
+use crate::protocol::wire_len::{kafka_compact_prefix, kafka_string_len};
/// An Iggy name caps at 255 bytes and a Kafka group's offset key is `kafka.cg.<group>`, so a
/// group id this gateway admits must leave room for that prefix (`docs/OFFSET_STORAGE.md`).
@@ -63,6 +71,20 @@ pub const fn is_valid_group_id(group_id: &str) -> bool {
/// one or two assignors.
pub const MAX_PROTOCOLS_PER_MEMBER: usize = 16;
+/// Legacy Kafka strings are an `i16` length. `DescribeGroups` v0-v4 and `ListGroups` v0-v2 encode
+/// `member_id` and (from v4) `group_instance_id` that way, and `kafka-protocol` refuses anything
+/// longer instead of writing it.
+const MAX_LEGACY_STRING_BYTES: usize = i16::MAX as usize;
+
+/// `protocol_type` is stored on the group and echoed by every `ListGroups` row. `i16::MAX` would
+/// let a few hundred pending-only joins pin 32 KiB types that sort ahead of real groups and fill
+/// the list response. 255 covers `consumer` and any stock protocol name.
+const MAX_PROTOCOL_TYPE_BYTES: usize = 255;
+
+/// `generate_member_id` appends `-{uuid}`. A UUID's display form is 36 bytes, so the prefix has
+/// to leave this much room or the id itself will not fit a legacy string.
+const MEMBER_ID_SUFFIX_BYTES: usize = 1 + 36;
+
/// Prefix for a generated member id when the client sent no `group_instance_id`. Kafka uses the
/// header's `client_id`, which handlers do not receive.
const DEFAULT_MEMBER_PREFIX: &str = "member";
@@ -81,6 +103,16 @@ pub enum Phase {
Stable,
}
+impl Phase {
+ const fn as_kafka_state(self) -> &'static str {
+ match self {
+ Self::PreparingRebalance => GROUP_STATE_PREPARING_REBALANCE,
+ Self::CompletingRebalance => GROUP_STATE_COMPLETING_REBALANCE,
+ Self::Stable => GROUP_STATE_STABLE,
+ }
+ }
+}
+
/// Outcome of one state-machine step: answer now, or park until `wake_at`.
pub enum Step<T> {
Respond(T),
@@ -182,6 +214,52 @@ impl Member {
.find(|(candidate, _)| candidate == name)
.map_or_else(Bytes::new, |(_, metadata)| metadata.clone())
}
+
+ fn metadata_len(&self, name: &StrBytes) -> usize {
+ self.protocols
+ .iter()
+ .find(|(candidate, _)| candidate == name)
+ .map_or(0, |(_, metadata)| metadata.len())
+ }
+
+ /// Encoded `DescribedGroupMember` size for `version`, without copying metadata or assignment.
+ /// `include_blobs` is false for Kafka's `summaryNoMetadata` shape: member id and instance id
+ /// stay, and metadata and assignment are empty.
+ fn described_len(
+ &self,
+ version: i16,
+ member_id: &StrBytes,
+ protocol_name: Option<&StrBytes>,
+ include_blobs: bool,
+ ) -> usize {
+ let flexible = version >= 5;
+ let mut total = kafka_string_len(flexible, member_id.len());
+ if version >= 4 {
+ total += kafka_nullable_string_len(
+ flexible,
+ self.group_instance_id.as_ref().map(|id| id.len()),
+ );
+ }
+ // `client_id` and `client_host` stay empty: the coordinator does not retain them.
+ total += kafka_string_len(flexible, 0);
+ total += kafka_string_len(flexible, 0);
+ let metadata_len = if include_blobs {
+ protocol_name.map_or(0, |name| self.metadata_len(name))
+ } else {
+ 0
+ };
+ let assignment_len = if include_blobs {
+ self.assignment.len()
+ } else {
+ 0
+ };
+ total += kafka_bytes_len(flexible, metadata_len);
+ total += kafka_bytes_len(flexible, assignment_len);
+ if flexible {
+ total += 1;
+ }
+ total
+ }
}
pub struct GroupState {
@@ -241,6 +319,26 @@ impl GroupState {
self.members.is_empty() && self.pending.is_empty()
}
+ /// Joined members, not pending ids. A v4+ first join holds only a pending id and Kafka
+ /// reports that group as `Empty`, which is what `--list --state Empty` filters on.
+ fn reported_state(&self) -> &'static str {
+ if self.members.is_empty() {
+ GROUP_STATE_EMPTY
+ } else {
+ self.phase.as_kafka_state()
+ }
+ }
+
+ /// Every stored deadline is already due. Pending ids count: a group still waiting on
+ /// `MEMBER_ID_REQUIRED` is visible until that id expires. An empty map is vacuously due,
+ /// which only matters for a group tick should already have removed.
+ fn deadlines_elapsed(&self, now: Instant) -> bool {
+ self.members
+ .values()
+ .all(|member| member.session_deadline <= now)
+ && self.pending.values().all(|deadline| *deadline <= now)
+ }
+
/// Would admitting `request` as `member_id` grow the leader's roster past its cap? The
/// member's own current entry is replaced, not added to.
fn roster_overflows(
@@ -629,6 +727,88 @@ impl GroupState {
self.sync_deadline = None;
self.bump();
}
+
+ /// `summary` forces Kafka's `summaryNoMetadata` shape even in `Stable`: empty `protocol_data`,
+ /// metadata and assignment. A group that is not `Stable` always uses that shape, because the
+ /// assignment still on the member is the previous generation and a rejoin has already
+ /// replaced the subscription.
+ fn description(&self, group_id: &StrBytes, summary: bool) -> GroupDescription {
+ let summary = summary || self.phase != Phase::Stable;
+ let protocol_name = if summary {
+ None
+ } else {
+ self.protocol_name.clone()
+ };
+ let members = self
+ .members
+ .iter()
+ .map(|(member_id, member)| MemberDescription {
+ member_id: member_id.clone(),
+ group_instance_id: member.group_instance_id.clone(),
+ metadata: if summary {
+ Bytes::new()
+ } else {
+ protocol_name
+ .as_ref()
+ .map_or_else(Bytes::new, |name| member.metadata_for(name))
+ },
+ assignment: if summary {
+ Bytes::new()
+ } else {
+ member.assignment.clone()
+ },
+ })
+ .collect();
+ GroupDescription {
+ group_id: group_id.clone(),
+ error: ERROR_NONE,
+ state: self.reported_state(),
+ protocol_type: self.protocol_type.clone(),
+ protocol_name,
+ members,
+ }
+ }
+
+ /// Encoded `DescribedGroup` size for `version`. Same fields [`Self::description`] copies,
+ /// priced from their lengths so the frame check does not have to build that snapshot first.
+ fn described_len(&self, version: i16, group_id: &StrBytes, summary: bool) -> usize {
+ let summary = summary || self.phase != Phase::Stable;
+ let flexible = version >= 5;
+ let mut total = 2;
+ if version >= 6 {
+ total += kafka_nullable_string_len(true, None);
+ }
+ total += kafka_string_len(flexible, group_id.len());
+ total += kafka_string_len(flexible, self.reported_state().len());
+ total += kafka_string_len(flexible, self.protocol_type.len());
+ let protocol_len = if summary {
+ 0
+ } else {
+ self.protocol_name.as_ref().map_or(0, |name| name.len())
+ };
+ total += kafka_string_len(flexible, protocol_len);
+ total += kafka_array_prefix(flexible, self.members.len());
+ let protocol_name = if summary {
+ None
+ } else {
+ self.protocol_name.as_ref()
+ };
+ for (member_id, member) in &self.members {
+ total = total.saturating_add(member.described_len(
+ version,
+ member_id,
+ protocol_name,
+ !summary,
+ ));
+ }
+ if version >= 3 {
+ total += 4;
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+ }
}
/// Tick every group and drop the ones that emptied, returning how many were reclaimed.
@@ -662,6 +842,8 @@ fn tick_group(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> bool {
};
group.tick(now);
if group.is_empty() {
+ // TODO(#3542): keep a group that only holds committed offsets and report Empty.
+ // Offsets are not stored here yet, so nothing remains that a later join would need.
groups.remove(group_id);
return false;
}
@@ -698,16 +880,18 @@ fn join_request_error(config: &GroupCoordinatorConfig, request: &JoinRequest) ->
if request.protocol_type.is_empty() || request.protocols.is_empty() {
return Some(ERROR_INCONSISTENT_GROUP_PROTOCOL);
}
- if request.protocols.len() > MAX_PROTOCOLS_PER_MEMBER {
+ if request.protocols.len() > MAX_PROTOCOLS_PER_MEMBER || legacy_strings_too_long(request) {
return Some(ERROR_INVALID_REQUEST);
}
// Names count too: they are retained alongside the metadata, and a request can carry many
- // long ones while declaring almost no metadata at all.
- let retained_bytes: usize = request
- .protocols
- .iter()
- .map(|(name, metadata)| name.len() + metadata.len())
- .sum::<usize>()
+ // long ones while declaring almost no metadata at all. `protocol_type` is retained on the
+ // group itself and echoed for every listed group, so it counts the same way.
+ let retained_bytes: usize = request.protocol_type.len()
+ + request
+ .protocols
+ .iter()
+ .map(|(name, metadata)| name.len() + metadata.len())
+ .sum::<usize>()
+ request
.group_instance_id
.as_ref()
@@ -718,6 +902,32 @@ fn join_request_error(config: &GroupCoordinatorConfig, request: &JoinRequest) ->
None
}
+/// A client-supplied `member_id`, `group_instance_id`, and each protocol `name` are echoed as
+/// legacy strings. `DescribeGroups` v0-v4 writes the selected protocol name as `protocol_data`
+/// with an i16 length. When the member id is generated, the instance id is the prefix and the
+/// `-{uuid}` suffix has to fit in the same cap. `protocol_type` is tighter than that cap: see
+/// [`MAX_PROTOCOL_TYPE_BYTES`].
+fn legacy_strings_too_long(request: &JoinRequest) -> bool {
+ if request.protocol_type.len() > MAX_PROTOCOL_TYPE_BYTES
+ || request.member_id.len() > MAX_LEGACY_STRING_BYTES
+ || request
+ .protocols
+ .iter()
+ .any(|(name, _)| name.len() > MAX_LEGACY_STRING_BYTES)
+ {
+ return true;
+ }
+ let Some(instance_id) = &request.group_instance_id else {
+ return false;
+ };
+ let limit = if request.member_id.is_empty() {
+ MAX_LEGACY_STRING_BYTES - MEMBER_ID_SUFFIX_BYTES
+ } else {
+ MAX_LEGACY_STRING_BYTES
+ };
+ instance_id.len() > limit
+}
+
/// Make sure `request.group_id` exists and is ticked, returning whether this call created it.
///
/// Reclamation runs only when the group cap is about to reject: walking every group on every
@@ -1122,6 +1332,247 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// How one requested id is answered. Priced before [`GroupState::description`] copies blobs.
+#[derive(Clone, Copy)]
+enum DescribedChoice {
+ /// Not in the map after the tick. Encoded as `Dead`.
+ Missing,
+ /// `summary` is Kafka's `summaryNoMetadata` shape.
+ Present { summary: bool },
+ /// Does not fit in the bytes still left in this response.
+ TooLarge,
+}
+
+/// One entry per distinct id, first-seen order, after that group's usual expiry tick.
+///
+/// Ids are deduped before the tick, so a request that repeats one group does not walk it once
+/// per copy. Each group is priced on its own. One that does not fit by itself is answered
+/// without member metadata. One that does not fit in what the earlier groups left is answered
+/// with [`ERROR_UNKNOWN_SERVER_ERROR`]. The connection is not closed: outbound frames are not
+/// capped, and closing makes `kafka-consumer-groups.sh --describe --all-groups` retry the same
+/// request until the client times out.
+pub fn describe_groups(
+ groups: &mut Groups,
+ group_ids: &[StrBytes],
+ version: i16,
+ max_frame_size: usize,
+ now: Instant,
+) -> Vec<Option<GroupDescription>> {
+ let group_ids = dedup_first_seen(group_ids);
+ for group_id in &group_ids {
+ tick_group(groups, group_id, now);
+ }
+ choose_described(version, groups, &group_ids, max_frame_size)
+ .into_iter()
+ .zip(group_ids)
+ .map(|(choice, group_id)| match choice {
+ DescribedChoice::Missing => None,
+ DescribedChoice::TooLarge => Some(too_large_description(&group_id)),
+ DescribedChoice::Present { summary } => groups
+ .get(&group_id)
+ .map(|group| group.description(&group_id, summary)),
+ })
+ .collect()
+}
+
+/// Groups currently in the map, ordered by group id.
+///
+/// Does not tick. Groups whose sessions and pending ids are already past `now` are skipped so
+/// the list agrees with a describe of the same id, which would tick them away and answer `Dead`.
+pub fn list_groups(groups: &Groups, now: Instant) -> Vec<GroupListing> {
+ let mut listed: Vec<GroupListing> = groups
+ .iter()
+ .filter(|(_, group)| !group.deadlines_elapsed(now))
+ .map(|(group_id, group)| GroupListing {
+ group_id: group_id.clone(),
+ protocol_type: group.protocol_type.clone(),
+ state: group.reported_state(),
+ })
+ .collect();
+ listed.sort_by(|left, right| left.group_id.cmp(&right.group_id));
+ listed
+}
+
+fn too_large_description(group_id: &StrBytes) -> GroupDescription {
+ GroupDescription {
+ group_id: group_id.clone(),
+ error: ERROR_UNKNOWN_SERVER_ERROR,
+ state: "",
+ protocol_type: StrBytes::from_static_str(""),
+ protocol_name: None,
+ members: Vec::new(),
+ }
+}
+
+fn choose_described(
+ version: i16,
+ groups: &Groups,
+ group_ids: &[StrBytes],
+ max_frame_size: usize,
+) -> Vec<DescribedChoice> {
+ let mut used = describe_fixed_len(version, group_ids.len());
+ let mut choices = Vec::with_capacity(group_ids.len());
+ for group_id in group_ids {
+ let Some(group) = groups.get(group_id) else {
+ let missing = missing_described_len(version, group_id);
+ if used.saturating_add(missing) <= max_frame_size {
+ used += missing;
+ choices.push(DescribedChoice::Missing);
+ } else {
+ used = used.saturating_add(too_large_described_len(version, group_id));
+ choices.push(DescribedChoice::TooLarge);
+ }
+ continue;
+ };
+ let natural_summary = group.phase != Phase::Stable;
+ let natural = group.described_len(version, group_id, natural_summary);
+ if used.saturating_add(natural) <= max_frame_size {
+ used += natural;
+ choices.push(DescribedChoice::Present {
+ summary: natural_summary,
+ });
+ continue;
+ }
+ // Full metadata does not fit in what is left. Use the summary shape only when the group
+ // would not fit even as the only group in the response. A group that fits alone and
+ // loses the rest of the batch gets an error, so a later describe of that id still
+ // returns the assignment.
+ if !natural_summary {
+ let alone =
+ describe_fixed_len(version, 1) + group.described_len(version, group_id, false);
+ if alone > max_frame_size {
+ let summary = group.described_len(version, group_id, true);
+ if used.saturating_add(summary) <= max_frame_size {
+ used += summary;
+ choices.push(DescribedChoice::Present { summary: true });
+ continue;
+ }
+ }
+ }
+ used = used.saturating_add(too_large_described_len(version, group_id));
+ choices.push(DescribedChoice::TooLarge);
+ }
+ choices
+}
+
+/// Header plus body for the choices [`choose_described`] would make. The 4-byte length prefix
+/// `send_response` writes is not part of that value. Test-only: production prices inside
+/// [`choose_described`] and does not need the summed total.
+#[cfg(test)]
+fn describe_frame_len(
+ version: i16,
+ groups: &Groups,
+ group_ids: &[StrBytes],
+ max_frame_size: usize,
+) -> usize {
+ let group_ids = dedup_first_seen(group_ids);
+ let mut total = describe_fixed_len(version, group_ids.len());
+ for (group_id, choice) in group_ids.iter().zip(choose_described(
+ version,
+ groups,
+ &group_ids,
+ max_frame_size,
+ )) {
+ let group_bytes = match choice {
+ DescribedChoice::Missing => missing_described_len(version, group_id),
+ DescribedChoice::TooLarge => too_large_described_len(version, group_id),
+ DescribedChoice::Present { summary } => groups
+ .get(group_id)
+ .map_or(0, |group| group.described_len(version, group_id, summary)),
+ };
+ total = total.saturating_add(group_bytes);
+ }
+ total
+}
+
+fn describe_fixed_len(version: i16, group_count: usize) -> usize {
+ let mut total = kafka_response_header_len(API_KEY_DESCRIBE_GROUPS, version);
+ if version >= 1 {
+ total += 4;
+ }
+ total += kafka_array_prefix(version >= 5, group_count);
+ if version >= 5 {
+ total += 1;
+ }
+ total
+}
+
+/// Per-group `UNKNOWN_SERVER_ERROR`. v6 carries [`DESCRIBE_RESPONSE_TOO_LARGE`]; older versions
+/// have no `error_message` field. Matches `described_group` in the handler.
+fn too_large_described_len(version: i16, group_id: &StrBytes) -> usize {
+ let flexible = version >= 5;
+ let mut total = 2;
+ if version >= 6 {
+ total += kafka_string_len(true, DESCRIBE_RESPONSE_TOO_LARGE.len());
+ }
+ total += kafka_string_len(flexible, group_id.len());
+ total += kafka_string_len(flexible, 0);
+ total += kafka_string_len(flexible, 0);
+ total += kafka_string_len(flexible, 0);
+ total += kafka_array_prefix(flexible, 0);
+ if version >= 3 {
+ total += 4;
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+}
+
+/// A group that is not in the map. v6 carries `GROUP_ID_NOT_FOUND` and `Group {id} not found.`;
+/// older versions encode error 0 and omit `error_message`. Matches `missing_group` in the handler.
+fn missing_described_len(version: i16, group_id: &StrBytes) -> usize {
+ let flexible = version >= 5;
+ let mut total = 2;
+ if version >= 6 {
+ let message_len = "Group ".len() + group_id.len() + " not found.".len();
+ total += kafka_string_len(true, message_len);
+ }
+ total += kafka_string_len(flexible, group_id.len());
+ total += kafka_string_len(flexible, "Dead".len());
+ total += kafka_string_len(flexible, 0);
+ total += kafka_string_len(flexible, 0);
+ total += kafka_array_prefix(flexible, 0);
+ if version >= 3 {
+ total += 4;
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+}
+
+fn kafka_response_header_len(api_key: i16, version: i16) -> usize {
+ if response_header_version(api_key, version) >= 1 {
+ 5
+ } else {
+ 4
+ }
+}
+
+fn kafka_nullable_string_len(flexible: bool, bytes: Option<usize>) -> usize {
+ match bytes {
+ Some(bytes) => kafka_string_len(flexible, bytes),
+ None if flexible => 1,
+ None => 2,
+ }
+}
+
+fn kafka_bytes_len(flexible: bool, bytes: usize) -> usize {
+ if flexible {
+ kafka_compact_prefix(bytes) + bytes
+ } else {
+ 4 + bytes
+ }
+}
+
+fn kafka_array_prefix(flexible: bool, count: usize) -> usize {
+ if flexible {
+ kafka_compact_prefix(count)
+ } else {
+ 4
+ }
+}
/// Kafka's `ClassicGroup.validateOffsetCommit`. Never creates a group.
///
/// A group with no members is Kafka's `Empty`: a negative generation commits without a member. A
@@ -1264,6 +1715,9 @@ fn generate_member_id(group_instance_id: Option<&StrBytes>) -> StrBytes {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::protocol::handlers::describe_groups::{
+ described_group, encode_response, missing_group,
+ };
const GROUP: &str = "g";
@@ -1376,6 +1830,22 @@ mod tests {
assert_ne!(error_of_or_none(&at_cap), Some(ERROR_INVALID_REQUEST));
}
+ /// A single protocol name past `MAX_LEGACY_STRING_BYTES` fits easily under the aggregate
+ /// `max_member_blob_bytes` cap (64 KiB), so only `legacy_strings_too_long` catches it -
+ /// otherwise `DescribeGroups` v0-v4 would later fail to encode this name as `protocol_data`
+ /// (an i16-length legacy string) and close the connection for an already-stored group.
+ #[test]
+ fn given_an_overlong_protocol_name_when_joining_should_answer_invalid_request() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let long_name = "p".repeat(MAX_LEGACY_STRING_BYTES + 1);
+
+ let over_limit = join_step(&mut groups, &config, &request("", &[&long_name]), now);
+ assert_eq!(error_of(&over_limit), ERROR_INVALID_REQUEST);
+ assert!(groups.is_empty(), "a rejected join must not create a group");
+ }
+
fn request(member_id: &str, protocols: &[&str]) -> JoinRequest {
request_with(member_id, protocols, 10, 5)
}
@@ -1642,6 +2112,359 @@ mod tests {
);
}
+ /// `protocol_type` is echoed on every list row. 255 is the cap; one byte over must not be
+ /// stored, and 255 itself still joins.
+ #[test]
+ fn given_a_protocol_type_past_255_bytes_when_joining_should_reject_it() {
+ let config = config();
+ let mut groups = Groups::new();
+ let mut request = request("", &["range"]);
+ request.protocol_type = StrBytes::from_string("p".repeat(MAX_PROTOCOL_TYPE_BYTES + 1));
+
+ let rejected = join_step(&mut groups, &config, &request, Instant::now());
+
+ assert_eq!(error_of(&rejected), ERROR_INVALID_REQUEST);
+ assert!(
+ groups.is_empty(),
+ "an overlong protocol_type must not be stored"
+ );
+
+ request.protocol_type = StrBytes::from_string("p".repeat(MAX_PROTOCOL_TYPE_BYTES));
+ let admitted = join_step(&mut Groups::new(), &config, &request, Instant::now());
+ assert_ne!(error_of(&admitted), ERROR_INVALID_REQUEST);
+ }
+
+ /// `protocol_type` is retained on the group, so it has to count toward `max_member_blob_bytes`.
+ /// The metadata below fits that cap on its own.
+ #[test]
+ fn given_a_protocol_type_that_overflows_the_retention_cap_when_joining_should_reject_it() {
+ let config = GroupCoordinatorConfig {
+ max_member_blob_bytes: 16,
+ ..config()
+ };
+ let mut groups = Groups::new();
+ let mut request = request("", &["range"]);
+ request.protocol_type = StrBytes::from_static_str("consumer");
+ request.protocols = vec![(StrBytes::from_static_str("range"), Bytes::from(vec![1; 10]))];
+
+ let step = join_step(&mut groups, &config, &request, Instant::now());
+
+ assert_eq!(error_of(&step), ERROR_INVALID_REQUEST);
+ assert!(groups.is_empty());
+ }
+
+ /// The generated member id is `{group_instance_id}-{uuid}`. The suffix has to be inside the
+ /// same legacy-string cap, or `DescribeGroups` v0-v4 cannot encode `member_id`.
+ #[test]
+ fn given_an_instance_id_that_leaves_no_room_for_the_member_suffix_when_joining_should_reject_it()
+ {
+ let config = config();
+ let mut groups = Groups::new();
+ let prefix_len = MAX_LEGACY_STRING_BYTES - MEMBER_ID_SUFFIX_BYTES;
+ let mut request = request("", &["range"]);
+ request.group_instance_id = Some(StrBytes::from_string("i".repeat(prefix_len + 1)));
+
+ let rejected = join_step(&mut groups, &config, &request, Instant::now());
+
+ assert_eq!(error_of(&rejected), ERROR_INVALID_REQUEST);
+ assert!(groups.is_empty());
+
+ request.group_instance_id = Some(StrBytes::from_string("i".repeat(prefix_len)));
+ let admitted = join_step(&mut Groups::new(), &config, &request, Instant::now());
+ assert_ne!(error_of_or_none(&admitted), Some(ERROR_INVALID_REQUEST));
+ assert_eq!(member_id_of(&admitted).len(), MAX_LEGACY_STRING_BYTES);
+ }
+
+ /// A client-supplied member id is not generated, so its own length is the cap. An instance
+ /// id that already has a member id is capped on its own, without reserving the suffix.
+ #[test]
+ fn given_a_member_id_past_the_legacy_string_cap_when_joining_should_reject_it() {
+ let config = config();
+ let mut groups = Groups::new();
+ let request = request(&"m".repeat(MAX_LEGACY_STRING_BYTES + 1), &["range"]);
+
+ let step = join_step(&mut groups, &config, &request, Instant::now());
+
+ assert_eq!(error_of(&step), ERROR_INVALID_REQUEST);
+ assert!(groups.is_empty());
+ }
+
+ #[test]
+ fn given_a_known_member_id_and_a_max_legacy_instance_id_when_joining_should_not_reserve_the_suffix()
+ {
+ let config = config();
+ let mut groups = Groups::new();
+ let mut request = request("known-member", &["range"]);
+ request.group_instance_id =
+ Some(StrBytes::from_string("i".repeat(MAX_LEGACY_STRING_BYTES)));
+
+ let step = join_step(&mut groups, &config, &request, Instant::now());
+
+ assert_eq!(error_of(&step), ERROR_UNKNOWN_MEMBER_ID);
+ }
+
+ /// Repeating an id must not build a second snapshot. A frame of 0 bytes answers that one
+ /// group with an error and leaves it in the map.
+ #[test]
+ fn given_a_repeated_group_id_when_describing_should_tick_and_snapshot_it_once() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let _ = join_step(&mut groups, &config, &request("", &["range"]), now);
+ let id = group_id();
+
+ let described = describe_groups(&mut groups, &[id.clone(), id.clone()], 0, usize::MAX, now);
+ assert_eq!(described.len(), 1);
+ assert_eq!(described[0].as_ref().unwrap().error, ERROR_NONE);
+
+ let refused = describe_groups(&mut groups, std::slice::from_ref(&id), 0, 0, now);
+ assert_eq!(
+ refused[0].as_ref().unwrap().error,
+ ERROR_UNKNOWN_SERVER_ERROR
+ );
+ assert!(
+ groups.contains_key(&id),
+ "a per-group size error must not drop the group"
+ );
+ }
+
+ /// The frame check has to match what `DescribeGroups` actually encodes, header included.
+ /// Fields are past the compact-varint boundary and the member has an instance id, which v4+
+ /// writes and v0-v3 does not.
+ #[test]
+ fn given_a_described_group_when_pricing_the_frame_should_match_the_encoder() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let joined = join_wide(&mut groups, &config, now);
+ let id = group_id();
+ let missing = StrBytes::from_static_str("missing");
+ assert_priced(&mut groups, &id, &missing, usize::MAX, now);
+
+ let synced = sync_step(
+ &mut groups,
+ &config,
+ &SyncRequest {
+ group_id: id.clone(),
+ generation_id: joined.generation_id,
+ member_id: joined.member_id.clone(),
+ protocol_type: None,
+ protocol_name: None,
+ assignments: vec![(joined.member_id, Bytes::from(vec![b'a'; 130]))],
+ },
+ now,
+ );
+ assert_eq!(error_of_sync(&synced), ERROR_NONE);
+ assert_eq!(groups[&id].phase, Phase::Stable);
+ assert_priced(&mut groups, &id, &missing, usize::MAX, now);
+ }
+
+ /// One stable group that does not fit on its own is summarized. The member stays, without
+ /// the blobs that made it too big.
+ #[test]
+ fn given_a_stable_group_one_byte_over_the_frame_when_describing_should_summarize() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ stabilize(&mut groups, &config, now, GROUP, &vec![b'x'; 400]);
+ let id = group_id();
+ let full = describe_frame_len(0, &groups, std::slice::from_ref(&id), usize::MAX);
+
+ let described = describe_groups(&mut groups, std::slice::from_ref(&id), 0, full - 1, now);
+ let group = described[0].as_ref().unwrap();
+ assert_eq!(group.error, ERROR_NONE);
+ assert_eq!(group.state, GROUP_STATE_STABLE);
+ assert!(group.protocol_name.is_none());
+ assert_eq!(group.members.len(), 1);
+ assert!(group.members[0].metadata.is_empty());
+ assert!(group.members[0].assignment.is_empty());
+ assert_priced_response(0, &mut groups, &[id], full - 1, now);
+ }
+
+ /// The second group fits on its own and does not fit beside the first, so it is an error
+ /// rather than a summary. Both groups stay.
+ #[test]
+ fn given_two_groups_past_the_batch_budget_when_describing_should_error_the_second() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let blob = vec![b'x'; 200];
+ stabilize(&mut groups, &config, now, "alpha", &blob);
+ stabilize(&mut groups, &config, now, "beta", &blob);
+ let first = StrBytes::from_static_str("alpha");
+ let second = StrBytes::from_static_str("beta");
+ let one = describe_frame_len(0, &groups, std::slice::from_ref(&first), usize::MAX);
+
+ let described = describe_groups(&mut groups, &[first.clone(), second.clone()], 0, one, now);
+ assert_eq!(described[0].as_ref().unwrap().error, ERROR_NONE);
+ assert!(
+ !described[0].as_ref().unwrap().members[0]
+ .assignment
+ .is_empty()
+ );
+ assert_eq!(
+ described[1].as_ref().unwrap().error,
+ ERROR_UNKNOWN_SERVER_ERROR
+ );
+ assert!(groups.contains_key(&first));
+ assert!(groups.contains_key(&second));
+ assert_priced_response(0, &mut groups, &[first, second], one, now);
+ }
+
+ /// A v4+ join that only reserved a member id has no joined members. Kafka calls that `Empty`.
+ #[test]
+ fn given_only_a_pending_member_when_describing_should_report_empty() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let mut request = request("", &["range"]);
+ request.require_known_member_id = true;
+ let step = join_step(&mut groups, &config, &request, now);
+ assert_eq!(error_of(&step), ERROR_MEMBER_ID_REQUIRED);
+
+ let listed = list_groups(&groups, now);
+ assert_eq!(listed.len(), 1);
+ assert_eq!(listed[0].state, GROUP_STATE_EMPTY);
+
+ let described = describe_groups(&mut groups, &[group_id()], 6, usize::MAX, now);
+ let group = described[0].as_ref().unwrap();
+ assert_eq!(group.error, ERROR_NONE);
+ assert_eq!(group.state, GROUP_STATE_EMPTY);
+ assert!(group.members.is_empty());
+ assert!(group.protocol_name.is_none());
+ assert!(groups.contains_key(&group_id()));
+ }
+
+ /// List skips a fully expired group and does not tick it. Describe then ticks and the group
+ /// is gone, which the handler reports as `Dead`.
+ #[test]
+ fn given_expired_sessions_when_listing_should_skip_without_ticking() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ stabilize(&mut groups, &config, now, GROUP, b"assign");
+ let id = group_id();
+ assert_eq!(list_groups(&groups, now).len(), 1);
+
+ let later = now + Duration::from_secs(11);
+ assert!(list_groups(&groups, later).is_empty());
+ assert_eq!(groups[&id].phase, Phase::Stable);
+ assert!(!groups[&id].members.is_empty());
+
+ let described =
+ describe_groups(&mut groups, std::slice::from_ref(&id), 6, usize::MAX, later);
+ assert!(described[0].is_none());
+ assert!(!groups.contains_key(&id));
+ }
+
+ fn assert_priced(
+ groups: &mut Groups,
+ id: &StrBytes,
+ missing: &StrBytes,
+ max_frame_size: usize,
+ now: Instant,
+ ) {
+ for version in 0..=6 {
+ for ids in [
+ vec![id.clone()],
+ vec![missing.clone()],
+ vec![id.clone(), missing.clone()],
+ ] {
+ assert_priced_response(version, groups, &ids, max_frame_size, now);
+ }
+ }
+ }
+
+ fn assert_priced_response(
+ version: i16,
+ groups: &mut Groups,
+ ids: &[StrBytes],
+ max_frame_size: usize,
+ now: Instant,
+ ) {
+ let priced = describe_frame_len(version, groups, ids, max_frame_size);
+ let described = describe_groups(groups, ids, version, max_frame_size, now);
+ let encoded = encode_described(version, &dedup_first_seen(ids), &described);
+ let header = kafka_response_header_len(API_KEY_DESCRIBE_GROUPS, version);
+ assert_eq!(
+ priced,
+ header + encoded.len(),
+ "version {version} ids {} priced {priced} encoded {}",
+ ids.len(),
+ header + encoded.len()
+ );
+ }
+
+ fn encode_described(
+ version: i16,
+ ids: &[StrBytes],
+ described: &[Option<GroupDescription>],
+ ) -> Bytes {
+ let groups = ids
+ .iter()
+ .zip(described)
+ .map(|(group_id, group)| {
+ group.clone().map_or_else(
+ || missing_group(version, group_id),
+ |group| described_group(version, group),
+ )
+ })
+ .collect();
+ encode_response(version, groups).expect("described group encodes")
+ }
+
+ fn join_wide(groups: &mut Groups, config: &GroupCoordinatorConfig, now: Instant) -> JoinResult {
+ let wide = "x".repeat(130);
+ let mut request = request("", &["range"]);
+ request.group_instance_id = Some(StrBytes::from_string(wide.clone()));
+ request.protocols = vec![(StrBytes::from_string(wide), Bytes::from(vec![b'm'; 130]))];
+ let step = join_step(groups, config, &request, now);
+ let Step::Respond(result) = step else {
+ panic!("an immediate join must answer");
+ };
+ assert_eq!(result.error, ERROR_NONE);
+ assert!(
+ result.member_id.len() >= 127,
+ "instance id plus the member suffix must cross the compact varint boundary"
+ );
+ result
+ }
+
+ fn stabilize(
+ groups: &mut Groups,
+ config: &GroupCoordinatorConfig,
+ now: Instant,
+ group: &str,
+ assignment: &[u8],
+ ) {
+ let step = join_step(groups, config, &request_for(group, ""), now);
+ let Step::Respond(joined) = step else {
+ panic!("an immediate join must answer");
+ };
+ assert_eq!(joined.error, ERROR_NONE, "{group}");
+ let synced = sync_step(
+ groups,
+ config,
+ &SyncRequest {
+ group_id: StrBytes::from_string(group.to_owned()),
+ generation_id: joined.generation_id,
+ member_id: joined.member_id.clone(),
+ protocol_type: None,
+ protocol_name: None,
+ assignments: vec![(joined.member_id, Bytes::copy_from_slice(assignment))],
+ },
+ now,
+ );
+ assert_eq!(error_of_sync(&synced), ERROR_NONE, "{group}");
+ }
+
+ fn error_of_sync(step: &Step<SyncResult>) -> i16 {
+ match step {
+ Step::Respond(result) => result.error,
+ Step::Wait { .. } => panic!("expected a sync response, not a park"),
+ }
+ }
+
/// The cap has to measure everything a member retains. Protocol names are kept alongside the
/// metadata, so counting metadata alone lets a request declare almost none while pinning
/// megabytes, which is what `max_member_blob_bytes` exists to prevent.
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index c7fc01244..9539773b8 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -40,8 +40,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, 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,
+ describe_groups, dispatch, fetch, find_coordinator, heartbeat, init_producer_id, join_group,
+ leave_group, list_groups, list_offsets, metadata, offset_commit, offset_fetch, produce,
+ respond_or_close, sync_group,
};
use crate::protocol::probe_board::ProbeBoard;
use crate::protocol::sasl::{
@@ -59,6 +60,8 @@ pub const API_KEY_JOIN_GROUP: i16 = 11;
pub const API_KEY_HEARTBEAT: i16 = 12;
pub const API_KEY_LEAVE_GROUP: i16 = 13;
pub const API_KEY_SYNC_GROUP: i16 = 14;
+pub const API_KEY_DESCRIBE_GROUPS: i16 = 15;
+pub const API_KEY_LIST_GROUPS: i16 = 16;
pub const API_KEY_SASL_HANDSHAKE: i16 = 17;
pub const API_KEY_API_VERSIONS: i16 = 18;
pub const API_KEY_CREATE_TOPICS: i16 = 19;
@@ -111,6 +114,12 @@ pub const ERROR_ILLEGAL_GENERATION: i16 = 22;
pub const ERROR_INCONSISTENT_GROUP_PROTOCOL: i16 = 23;
pub const ERROR_INVALID_GROUP_ID: i16 = 24;
pub const ERROR_UNKNOWN_MEMBER_ID: i16 = 25;
+/// `DescribeGroups` v6: the named group is not on this coordinator.
+///
+/// Earlier versions of the same response report that miss as state `Dead` with error 0.
+/// Kafka 4.0's `describeGroups` splits it the same way, and a v6 client treats error 0 as a
+/// group that exists.
+pub const ERROR_GROUP_ID_NOT_FOUND: i16 = ResponseError::GroupIdNotFound.code();
pub const ERROR_INVALID_SESSION_TIMEOUT: i16 = 26;
/// How a follower learns to rejoin: its heartbeat is answered with this while the group prepares.
pub const ERROR_REBALANCE_IN_PROGRESS: i16 = 27;
@@ -295,6 +304,8 @@ static SUPPORTED_RANGES: &[ApiVersionRange] = &[
heartbeat::RANGE,
leave_group::RANGE,
sync_group::RANGE,
+ describe_groups::RANGE,
+ list_groups::RANGE,
];
#[must_use]
diff --git a/gateways/kafka/src/protocol/bounds_guard.rs b/gateways/kafka/src/protocol/bounds_guard.rs
index 90c711fbf..0d376d987 100644
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@ -1077,7 +1077,9 @@ pub fn validate_sync_group_shape(version: i16, body: &Bytes, max_frame_size: usi
Ok(())
}
-/// Mirrors the field order `OffsetCommitRequest::decode` walks.
+/// Mirrors the field order `DescribeGroupsRequest::decode` , `OffsetCommitRequest::decode` walks.
+///
+/// Group ids are echoed, so their lengths count toward the projected response.
///
/// # Errors
///
@@ -1203,6 +1205,70 @@ pub fn validate_offset_fetch_shape(
Ok(())
}
+/// Mirrors the field order `DescribeGroupsRequest::decode` walks.
+///
+/// Group ids are echoed, so their lengths count toward the projected response.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string 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_describe_groups_shape(
+ version: i16,
+ body: &Bytes,
+ max_frame_size: usize,
+) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ let flexible = version >= 5;
+ let groups_count = if flexible {
+ c.compact_array_count()?
+ } else {
+ c.legacy_array_count()?
+ };
+ for _ in 0..groups_count {
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ }
+ if version >= 3 {
+ let _include_authorized_operations = c.read_bool()?;
+ }
+ if flexible {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+/// Mirrors the field order `ListGroupsRequest::decode` walks.
+///
+/// The response is the coordinator's group list, not an echo of this body, so filter strings are
+/// walked only to reject a length that cannot fit the frame.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string 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_list_groups_shape(version: i16, body: &Bytes, max_frame_size: usize) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ if version >= 4 {
+ let states_count = c.compact_array_count()?;
+ for _ in 0..states_count {
+ c.compact_string(false)?;
+ }
+ }
+ if version >= 5 {
+ let types_count = c.compact_array_count()?;
+ for _ in 0..types_count {
+ c.compact_string(false)?;
+ }
+ }
+ if version >= 3 {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+
/// Cap on a whole `SaslAuthenticate` body.
///
/// The body is one length-prefixed blob, so capping it is the same as capping `auth_bytes`.
@@ -1420,7 +1486,10 @@ fn walk_offset_fetch_topics(
#[cfg(test)]
mod tests {
use bytes::BytesMut;
- use kafka_protocol::messages::{LeaveGroupRequest, OffsetCommitRequest, OffsetFetchRequest};
+ use kafka_protocol::messages::{
+ DescribeGroupsRequest, LeaveGroupRequest, ListGroupsRequest, OffsetCommitRequest,
+ OffsetFetchRequest,
+ };
use super::*;
use crate::protocol::handlers::decode_exhaustive;
@@ -1984,6 +2053,58 @@ mod tests {
assert!(validate_alter_configs_shape(2, &body, TEST_MAX_FRAME_SIZE).is_err());
}
+ #[test]
+ fn describe_groups_v5_huge_group_count_rejected() {
+ let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]);
+ assert!(validate_describe_groups_shape(5, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn describe_groups_v0_one_group_accepted() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x01, // groups: 1
+ 0x00, 0x01, b'g', // group id
+ ]);
+ assert!(validate_describe_groups_shape(0, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<DescribeGroupsRequest>(0, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn describe_groups_v5_one_group_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, // groups: 1
+ 0x02, b'g', // group id
+ 0x00, // include_authorized_operations
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_describe_groups_shape(5, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<DescribeGroupsRequest>(5, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn list_groups_v0_empty_body_accepted() {
+ let body = Bytes::new();
+ assert!(validate_list_groups_shape(0, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<ListGroupsRequest>(0, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn list_groups_v5_empty_filters_accepted() {
+ let body = Bytes::from_static(&[
+ 0x01, // states_filter: empty
+ 0x01, // types_filter: empty
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_list_groups_shape(5, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<ListGroupsRequest>(5, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn list_groups_v4_huge_states_filter_rejected() {
+ let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]);
+ assert!(validate_list_groups_shape(4, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
#[test]
fn offset_commit_v2_null_metadata_accepted() {
let body = Bytes::from_static(&[
diff --git a/gateways/kafka/src/protocol/dedup.rs b/gateways/kafka/src/protocol/dedup.rs
new file mode 100644
index 000000000..ce2e2d42b
--- /dev/null
+++ b/gateways/kafka/src/protocol/dedup.rs
@@ -0,0 +1,73 @@
+// 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.
+
+//! Shared first-seen-order deduplication for Kafka name lists.
+//!
+//! `Metadata`'s requested topic names and `DescribeGroups`' requested group ids both need to
+//! collapse a repeated name to one answer rather than multiplying the encoded response by the
+//! repeat count. Kept in one place so the two copies cannot drift apart.
+
+use std::collections::HashSet;
+
+use kafka_protocol::protocol::StrBytes;
+
+/// `names` with every repeat after the first dropped, in first-seen order.
+///
+/// Borrows into the dedup set rather than cloning into it; only a kept name's single `clone`
+/// (into the result) ever runs.
+#[must_use]
+pub fn dedup_first_seen(names: &[StrBytes]) -> Vec<StrBytes> {
+ let mut seen = HashSet::with_capacity(names.len());
+ let mut distinct = Vec::with_capacity(names.len());
+ for name in names {
+ if seen.insert(name.as_str()) {
+ distinct.push(name.clone());
+ }
+ }
+ distinct
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn dedup_first_seen_finds_a_name_repeated_across_the_list() {
+ let names = [
+ StrBytes::from_static_str("orders"),
+ StrBytes::from_static_str("payments"),
+ StrBytes::from_static_str("orders"),
+ ];
+ let distinct = dedup_first_seen(&names);
+ assert_eq!(
+ distinct,
+ vec![
+ StrBytes::from_static_str("orders"),
+ StrBytes::from_static_str("payments"),
+ ]
+ );
+ }
+
+ #[test]
+ fn dedup_first_seen_is_unchanged_when_every_name_is_unique() {
+ let names = [
+ StrBytes::from_static_str("orders"),
+ StrBytes::from_static_str("payments"),
+ ];
+ assert_eq!(dedup_first_seen(&names), names.to_vec());
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/describe_groups.rs b/gateways/kafka/src/protocol/handlers/describe_groups.rs
new file mode 100644
index 000000000..e8f59fd57
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs
@@ -0,0 +1,165 @@
+// 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.
+
+//! `DescribeGroups` (API key 15).
+//!
+//! Reports the members, assignment and state of groups this coordinator holds. A named group
+//! that is not here is `Dead`: v6 also carries `GROUP_ID_NOT_FOUND`, and earlier versions do
+//! not, matching Kafka 4.0's `describeGroups`. A v6 client treats error 0 as a real group.
+//!
+//! `include_authorized_operations` is not answered. The field stays at the omitted sentinel
+//! (`i32::MIN`): there is no group ACL bitfield to report, and any other value fails the
+//! encoder below v3.
+
+use bytes::Bytes;
+use kafka_protocol::messages::describe_groups_response::{DescribedGroup, DescribedGroupMember};
+use kafka_protocol::messages::{DescribeGroupsRequest, DescribeGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::error::Result;
+use crate::group::{DESCRIBE_RESPONSE_TOO_LARGE, GroupDescription, MemberDescription};
+use crate::protocol::api::{
+ API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ERROR_GROUP_ID_NOT_FOUND, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_describe_groups_shape;
+use crate::protocol::dedup::dedup_first_seen;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, respond_or_close, unsupported_version_response,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DESCRIBE_GROUPS,
+ min_version: 0,
+ max_version: 6,
+};
+
+/// Kafka's `Dead` state. This coordinator never stores it: an absent group is reported as one.
+const DEAD: &str = "Dead";
+/// First version whose response has `error_message`, and the first on which Kafka 4.0 sends
+/// `GROUP_ID_NOT_FOUND` for a missing group.
+const FIRST_ERROR_MESSAGE_VERSION: i16 = 6;
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DESCRIBE_GROUPS, api_version) {
+ // No top-level error field to carry a code on, so this answers with an empty group
+ // list rather than a dedicated error response. Versions outside 0-6 fail in the
+ // encoder, which is what makes `unsupported_version_response` close the connection
+ // instead - a version this encoder accepts is not a version the firewall rejects.
+ return unsupported_version_response(API_KEY_DESCRIBE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request =
+ match decode_guarded::<DescribeGroupsRequest>(api_version, body, |version, body| {
+ validate_describe_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ // The response has no top-level error field, and a failed decode yields no group
+ // id to hang `INVALID_REQUEST` on. An empty group list would look like success.
+ tracing::debug!(%error, api_version, "Failed to decode DescribeGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ // One result per distinct id. Repeating one group would otherwise copy its members into the
+ // response once per copy, and the shape guard only charges the id strings.
+ let group_ids = dedup_first_seen(&requested);
+ let described = state
+ .groups
+ .describe_groups(&group_ids, api_version, state.max_frame_size)
+ .await;
+ let groups = group_ids
+ .iter()
+ .zip(described)
+ .map(|(group_id, group)| {
+ group.map_or_else(
+ || missing_group(api_version, group_id),
+ |group| described_group(api_version, group),
+ )
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, groups), "DescribeGroups")
+}
+
+pub(crate) fn described_group(version: i16, group: GroupDescription) -> DescribedGroup {
+ if group.error != ERROR_NONE {
+ return too_large_group(version, &group.group_id);
+ }
+ DescribedGroup::default()
+ .with_group_id(GroupId(group.group_id))
+ .with_group_state(StrBytes::from_static_str(group.state))
+ .with_protocol_type(group.protocol_type)
+ .with_protocol_data(group.protocol_name.unwrap_or_default())
+ .with_members(group.members.into_iter().map(described_member).collect())
+}
+
+/// Empty member list and empty strings, matching Kafka's `DescribeGroupsResponse.groupError`.
+/// The code is not retried by the admin tool, unlike closing the connection.
+fn too_large_group(version: i16, group_id: &StrBytes) -> DescribedGroup {
+ let mut group = DescribedGroup::default()
+ .with_error_code(ERROR_UNKNOWN_SERVER_ERROR)
+ .with_group_id(GroupId(group_id.clone()))
+ .with_group_state(StrBytes::from_static_str(""))
+ .with_protocol_type(StrBytes::from_static_str(""))
+ .with_protocol_data(StrBytes::from_static_str(""));
+ if version >= FIRST_ERROR_MESSAGE_VERSION {
+ group =
+ group.with_error_message(Some(StrBytes::from_static_str(DESCRIBE_RESPONSE_TOO_LARGE)));
+ }
+ group
+}
+
+fn described_member(member: MemberDescription) -> DescribedGroupMember {
+ // `client_id` and `client_host` stay empty: the coordinator does not retain the request
+ // header's client id, and handlers are not given the connection's peer address.
+ DescribedGroupMember::default()
+ .with_member_id(member.member_id)
+ .with_group_instance_id(member.group_instance_id)
+ .with_member_metadata(member.metadata)
+ .with_member_assignment(member.assignment)
+}
+
+pub(crate) fn missing_group(version: i16, group_id: &StrBytes) -> DescribedGroup {
+ let mut group = DescribedGroup::default()
+ .with_group_id(GroupId(group_id.clone()))
+ .with_group_state(StrBytes::from_static_str(DEAD));
+ if version >= FIRST_ERROR_MESSAGE_VERSION {
+ group = group
+ .with_error_code(ERROR_GROUP_ID_NOT_FOUND)
+ .with_error_message(Some(StrBytes::from_string(format!(
+ "Group {} not found.",
+ group_id.as_str()
+ ))));
+ }
+ group
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, groups: Vec<DescribedGroup>) -> Result<Bytes> {
+ let response = DescribeGroupsResponse::default().with_groups(groups);
+ encode_message(&response, version, 64)
+}
diff --git a/gateways/kafka/src/protocol/handlers/list_groups.rs b/gateways/kafka/src/protocol/handlers/list_groups.rs
new file mode 100644
index 000000000..bf29264cc
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/list_groups.rs
@@ -0,0 +1,246 @@
+// 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.
+
+//! `ListGroups` (API key 16).
+//!
+//! Lists the groups this coordinator holds. v4+ `states_filter` and v5 `types_filter` keep a
+//! group only when the filter is empty or one entry matches, case-insensitively, after trim.
+//! A filter value this coordinator does not use matches nothing, so the group is left out and
+//! the response error stays 0. Kafka 4.0's `listGroups` does the same ("if invalid, no groups
+//! are returned").
+
+use bytes::Bytes;
+use kafka_protocol::messages::list_groups_response::ListedGroup;
+use kafka_protocol::messages::{GroupId, ListGroupsRequest, ListGroupsResponse};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::error::Result;
+use crate::group::GroupListing;
+use crate::protocol::api::{
+ API_KEY_LIST_GROUPS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_list_groups_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, respond_or_close, unsupported_version_response,
+};
+use crate::protocol::header::response_header_version;
+use crate::protocol::wire_len::{kafka_compact_prefix, kafka_string_len};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_LIST_GROUPS,
+ min_version: 0,
+ max_version: 5,
+};
+
+/// `ListGroups` v5 `group_type` for the classic protocol this coordinator runs.
+const CLASSIC_GROUP_TYPE: &str = "classic";
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ if !is_supported_version(API_KEY_LIST_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_LIST_GROUPS, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let request = match decode_guarded::<ListGroupsRequest>(api_version, body, |version, body| {
+ validate_list_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode ListGroups request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "ListGroups",
+ );
+ }
+ };
+
+ let listings: Vec<GroupListing> = state
+ .groups
+ .list_groups()
+ .await
+ .into_iter()
+ .filter(|group| {
+ selected(&request.states_filter, group.state)
+ && selected(&request.types_filter, CLASSIC_GROUP_TYPE)
+ })
+ .collect();
+ // `protocol_type` is echoed for every group. The shape guard never sees that list, so a
+ // coordinator holding many long types can still outgrow the frame - truncate rather than
+ // close, since this is server-sized state, not a request the client chose (same approach as
+ // `metadata::truncate_all_topics_to_frame_budget` for the identical problem).
+ let listings = truncate_listings_to_frame_budget(api_version, listings, state.max_frame_size);
+ let groups = listings.into_iter().map(listed_group).collect();
+ respond_or_close(encode_response(api_version, groups), "ListGroups")
+}
+
+fn selected(filter: &[StrBytes], value: &str) -> bool {
+ filter.is_empty()
+ || filter
+ .iter()
+ .any(|wanted| wanted.as_str().trim().eq_ignore_ascii_case(value))
+}
+
+/// Keeps as many listed groups as fit `max_frame_size`, dropping the rest rather than closing
+/// the connection - same approach as `metadata::truncate_all_topics_to_frame_budget` for the
+/// identical problem: this coordinator's own state, not a request the client chose, is what
+/// grew the response past the budget.
+fn truncate_listings_to_frame_budget(
+ version: i16,
+ mut listings: Vec<GroupListing>,
+ max_frame_size: usize,
+) -> Vec<GroupListing> {
+ let mut cumulative_bytes = list_response_frame_len(version, &[]);
+ let mut keep = listings.len();
+ for (index, group) in listings.iter().enumerate() {
+ let next = cumulative_bytes + listed_group_len(version, group);
+ if next > max_frame_size {
+ keep = index;
+ break;
+ }
+ cumulative_bytes = next;
+ }
+ if keep < listings.len() {
+ tracing::warn!(
+ total_groups = listings.len(),
+ kept_groups = keep,
+ max_frame_size,
+ "ListGroups response would exceed max_frame_size; truncating rather than closing \
+ the connection"
+ );
+ listings.truncate(keep);
+ }
+ listings
+}
+
+/// Header plus body, the length `send_response` writes and `read_frame` compares to
+/// `max_frame_size`.
+fn list_response_frame_len(version: i16, groups: &[GroupListing]) -> usize {
+ let flexible = version >= 3;
+ let mut total = if response_header_version(API_KEY_LIST_GROUPS, version) >= 1 {
+ 5
+ } else {
+ 4
+ };
+ if version >= 1 {
+ total += 4;
+ }
+ total += 2;
+ total += if flexible {
+ kafka_compact_prefix(groups.len())
+ } else {
+ 4
+ };
+ for group in groups {
+ total = total.saturating_add(listed_group_len(version, group));
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+}
+
+fn listed_group_len(version: i16, group: &GroupListing) -> usize {
+ let flexible = version >= 3;
+ let mut total = kafka_string_len(flexible, group.group_id.len());
+ total += kafka_string_len(flexible, group.protocol_type.len());
+ if version >= 4 {
+ total += kafka_string_len(true, group.state.len());
+ }
+ if version >= 5 {
+ total += kafka_string_len(true, CLASSIC_GROUP_TYPE.len());
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+}
+
+fn listed_group(group: GroupListing) -> ListedGroup {
+ ListedGroup::default()
+ .with_group_id(GroupId(group.group_id))
+ .with_protocol_type(group.protocol_type)
+ .with_group_state(StrBytes::from_static_str(group.state))
+ .with_group_type(StrBytes::from_static_str(CLASSIC_GROUP_TYPE))
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, groups: Vec<ListedGroup>) -> Result<Bytes> {
+ let response = ListGroupsResponse::default().with_groups(groups);
+ encode_message(&response, version, 64)
+}
+
+/// # 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> {
+ let response = ListGroupsResponse::default().with_error_code(error_code);
+ encode_message(&response, version, 16)
+}
+
+#[cfg(test)]
+mod tests {
+ use kafka_protocol::protocol::StrBytes;
+
+ use super::{
+ API_KEY_LIST_GROUPS, GroupListing, encode_response, list_response_frame_len, listed_group,
+ };
+ use crate::protocol::header::response_header_version;
+
+ /// v0-v2 header is 4 bytes, v3+ is 5. v4 adds the state string and v3+ adds a tag byte.
+ /// Group id and protocol type sit past the compact-varint boundary so a 1-byte length
+ /// prefix cannot accidentally match.
+ #[test]
+ fn given_listings_when_pricing_v0_through_v5_should_match_the_encoder() {
+ let group_id = "g".repeat(200);
+ let protocol_type = "p".repeat(200);
+ let listings = vec![
+ GroupListing {
+ group_id: StrBytes::from_string(group_id.clone()),
+ protocol_type: StrBytes::from_string(protocol_type.clone()),
+ state: "CompletingRebalance",
+ },
+ GroupListing {
+ group_id: StrBytes::from_string(format!("!{group_id}")),
+ protocol_type: StrBytes::from_string(protocol_type),
+ state: "Empty",
+ },
+ ];
+ for version in 0..=5 {
+ let priced = list_response_frame_len(version, &listings);
+ let encoded = encode_response(
+ version,
+ listings.iter().cloned().map(listed_group).collect(),
+ )
+ .expect("list encodes");
+ let header = if response_header_version(API_KEY_LIST_GROUPS, version) >= 1 {
+ 5
+ } else {
+ 4
+ };
+ assert_eq!(
+ priced,
+ header + encoded.len(),
+ "version {version} priced {priced} encoded {}",
+ header + encoded.len()
+ );
+ }
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/metadata.rs b/gateways/kafka/src/protocol/handlers/metadata.rs
index bfa6aea7f..0b1742cd8 100644
--- a/gateways/kafka/src/protocol/handlers/metadata.rs
+++ b/gateways/kafka/src/protocol/handlers/metadata.rs
@@ -17,8 +17,6 @@
//! Metadata (API key 3).
-use std::collections::HashSet;
-
use bytes::Bytes;
use kafka_protocol::messages::metadata_response::{
MetadataResponseBroker, MetadataResponsePartition, MetadataResponseTopic,
@@ -34,6 +32,7 @@ use crate::protocol::api::{
is_supported_version, supported_max_version,
};
use crate::protocol::bounds_guard::validate_metadata_shape;
+use crate::protocol::dedup::dedup_first_seen;
use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
pub const RANGE: ApiVersionRange = ApiVersionRange {
@@ -194,7 +193,7 @@ async fn resolve_requested_named_topics(
bridge: &IggyBridge,
names: &[StrBytes],
) -> Vec<TopicResult> {
- let distinct = dedup_topic_names(names);
+ let distinct = dedup_first_seen(names);
match tokio::time::timeout(REQUEST_DEADLINE, bridge.get_kafka_topics(&distinct)).await {
Ok(Ok(resolved)) => resolved
.into_iter()
@@ -355,17 +354,6 @@ const fn error_result(name: StrBytes, error_code: i16) -> TopicResult {
/// Drops repeats, keeping first-seen order so a capped or timed-out response still answers a
/// deterministic prefix of the request rather than an arbitrary hash-order subset.
-fn dedup_topic_names(names: &[StrBytes]) -> Vec<StrBytes> {
- let mut seen = HashSet::with_capacity(names.len());
- let mut distinct = Vec::with_capacity(names.len());
- for name in names {
- if seen.insert(name.as_str()) {
- distinct.push(name.clone());
- }
- }
- distinct
-}
-
/// # Errors
///
/// Returns an error when `kafka_protocol` cannot encode the response at `response_version`.
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 499b497f4..2e2282229 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -29,12 +29,14 @@ pub mod api_versions;
pub mod create_topics;
pub mod delete_topics;
pub mod describe_configs;
+pub mod describe_groups;
pub mod fetch;
pub mod find_coordinator;
pub mod heartbeat;
pub mod init_producer_id;
pub mod join_group;
pub mod leave_group;
+pub mod list_groups;
pub mod list_offsets;
pub mod metadata;
pub mod offset_commit;
@@ -57,11 +59,12 @@ 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_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,
+ API_KEY_DESCRIBE_CONFIGS, API_KEY_DESCRIBE_GROUPS, 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_GROUPS, 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.
@@ -124,6 +127,8 @@ pub async fn dispatch(
API_KEY_HEARTBEAT => heartbeat::handle(state, connection, api_version, body).await,
API_KEY_LEAVE_GROUP => leave_group::handle(state, api_version, body).await,
API_KEY_SYNC_GROUP => sync_group::handle(state, api_version, body).await,
+ API_KEY_DESCRIBE_GROUPS => describe_groups::handle(state, api_version, body).await,
+ API_KEY_LIST_GROUPS => list_groups::handle(state, api_version, body).await,
API_KEY_INIT_PRODUCER_ID => init_producer_id::handle(state, api_version, body).await,
_ => HandleOutcome::Close,
}
diff --git a/gateways/kafka/src/protocol/mod.rs b/gateways/kafka/src/protocol/mod.rs
index a32b2d092..1ade5911f 100644
--- a/gateways/kafka/src/protocol/mod.rs
+++ b/gateways/kafka/src/protocol/mod.rs
@@ -18,7 +18,9 @@
pub mod acl;
pub mod api;
pub mod bounds_guard;
+pub mod dedup;
pub mod handlers;
pub mod header;
pub(crate) mod probe_board;
pub mod sasl;
+pub mod wire_len;
diff --git a/gateways/kafka/src/protocol/wire_len.rs b/gateways/kafka/src/protocol/wire_len.rs
new file mode 100644
index 000000000..3602273f0
--- /dev/null
+++ b/gateways/kafka/src/protocol/wire_len.rs
@@ -0,0 +1,83 @@
+// 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.
+
+//! Shared Kafka wire-size pricing primitives.
+//!
+//! `ListGroups`' hand-computed response pricing and the consumer-group coordinator's own
+//! stored-state pricing both estimate encoded field sizes before `kafka_protocol` ever builds a
+//! response, so a frame that would exceed `max_frame_size` can be caught up front. Kept in one
+//! place so the two copies cannot drift out of sync with each other or with `kafka_protocol`'s
+//! actual encoder arithmetic.
+
+/// Length, in bytes, of the unsigned varint `kafka_protocol` writes for every compact-encoding
+/// length prefix.
+#[must_use]
+pub const fn unsigned_varint_len(value: u32) -> usize {
+ match value {
+ 0x00..=0x7f => 1,
+ 0x80..=0x3fff => 2,
+ 0x4000..=0x001f_ffff => 3,
+ 0x0020_0000..=0x0fff_ffff => 4,
+ _ => 5,
+ }
+}
+
+/// Length prefix of a compact string, compact bytes field, or compact array: the unsigned
+/// varint of `len + 1`.
+#[must_use]
+pub fn kafka_compact_prefix(len: usize) -> usize {
+ let wire = u32::try_from(len.saturating_add(1)).unwrap_or(u32::MAX);
+ unsigned_varint_len(wire)
+}
+
+/// A legacy (2-byte length) or compact (varint length) string field's total wire size.
+#[must_use]
+pub fn kafka_string_len(flexible: bool, bytes: usize) -> usize {
+ if flexible {
+ kafka_compact_prefix(bytes) + bytes
+ } else {
+ 2 + bytes
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn unsigned_varint_len_matches_kafka_protocols_own_boundaries() {
+ assert_eq!(unsigned_varint_len(0), 1);
+ assert_eq!(unsigned_varint_len(0x7f), 1);
+ assert_eq!(unsigned_varint_len(0x80), 2);
+ assert_eq!(unsigned_varint_len(0x3fff), 2);
+ assert_eq!(unsigned_varint_len(0x4000), 3);
+ assert_eq!(unsigned_varint_len(u32::MAX), 5);
+ }
+
+ #[test]
+ fn kafka_compact_prefix_prices_len_plus_one() {
+ assert_eq!(kafka_compact_prefix(0), 1); // N+1 = 1
+ assert_eq!(kafka_compact_prefix(126), 1); // N+1 = 127
+ assert_eq!(kafka_compact_prefix(127), 2); // N+1 = 128
+ }
+
+ #[test]
+ fn kafka_string_len_differs_between_flexible_and_legacy() {
+ assert_eq!(kafka_string_len(false, 10), 12); // 2-byte length + bytes
+ assert_eq!(kafka_string_len(true, 10), 11); // 1-byte compact prefix + bytes
+ }
+}
diff --git a/gateways/kafka/tests/common/scope.rs b/gateways/kafka/tests/common/scope.rs
index 63b4f38b9..b263570cf 100644
--- a/gateways/kafka/tests/common/scope.rs
+++ b/gateways/kafka/tests/common/scope.rs
@@ -37,6 +37,8 @@ pub const SCOPED_API_KEYS: &[(i16, &str, i16, i16)] = &[
(12, "Heartbeat", 0, 4),
(13, "LeaveGroup", 0, 5),
(14, "SyncGroup", 0, 5),
+ (15, "DescribeGroups", 0, 6),
+ (16, "ListGroups", 0, 5),
(18, "ApiVersions", 0, 3),
(19, "CreateTopics", 2, 5),
(20, "DeleteTopics", 1, 5),
diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs
index 14c56db76..9add5afa3 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -45,7 +45,11 @@ 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")];
+pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[
+ (21, "DeleteRecords"),
+ (42, "DeleteGroups"),
+ (44, "IncrementalAlterConfigs"),
+];
/// Append Metadata request fields that follow the topics array for `version`.
fn write_metadata_request_trailer(enc: &mut Encoder, version: i16) {
@@ -547,7 +551,49 @@ pub fn build_create_topics_request_with_sections(version: i16, topic: &str) -> B
enc.freeze()
}
-// ── Consumer group coordination (keys 10, 11, 12, 14) ───────────────────────
+// ── Consumer group coordination (keys 10-16) ────────────────────────────────
+
+/// `DescribeGroups` request naming `groups`. `include_authorized_operations` is written from v3.
+pub fn build_describe_groups_request(
+ version: i16,
+ groups: &[&str],
+ include_authorized_operations: bool,
+) -> Bytes {
+ let flexible = version >= 5;
+ let mut enc = Encoder::with_capacity(64);
+ write_array_count(&mut enc, flexible, groups.len());
+ for group in groups {
+ write_string(&mut enc, flexible, Some(group));
+ }
+ if version >= 3 {
+ enc.write_bool(include_authorized_operations);
+ }
+ if flexible {
+ enc.write_empty_tagged_fields();
+ }
+ enc.freeze()
+}
+
+/// `ListGroups` request. `states` is written from v4 and `group_types` from v5.
+pub fn build_list_groups_request(version: i16, states: &[&str], group_types: &[&str]) -> Bytes {
+ let mut enc = Encoder::with_capacity(64);
+ if version >= 4 {
+ enc.write_varint((states.len() + 1) as u64);
+ for state in states {
+ enc.write_compact_nullable_string(Some(state));
+ }
+ }
+ if version >= 5 {
+ enc.write_varint((group_types.len() + 1) as u64);
+ for group_type in group_types {
+ enc.write_compact_nullable_string(Some(group_type));
+ }
+ }
+ if version >= 3 {
+ enc.write_empty_tagged_fields();
+ }
+ enc.freeze()
+}
/// Write a Kafka string, compact or legacy by `flexible`.
fn write_string(enc: &mut Encoder, flexible: bool, value: Option<&str>) {
diff --git a/gateways/kafka/tests/consumer_group_tests.rs b/gateways/kafka/tests/consumer_group_tests.rs
index 89cc7087c..30af75466 100644
--- a/gateways/kafka/tests/consumer_group_tests.rs
+++ b/gateways/kafka/tests/consumer_group_tests.rs
@@ -16,7 +16,7 @@
// under the License.
//! Consumer group coordination: `FindCoordinator`, `JoinGroup`, Heartbeat, `LeaveGroup`,
-//! `SyncGroup`.
+//! `SyncGroup`, `DescribeGroups`, `ListGroups`.
//!
//! Requests go through `handle_request_bounded` against one shared `GatewayState`, because
//! `handle_request` builds a fresh coordinator per call and no two requests would ever see the
@@ -48,20 +48,23 @@ use tokio_util::sync::CancellationToken;
use iggy_gateway_kafka::GatewayConfig;
use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig, SyncRequest};
use iggy_gateway_kafka::protocol::api::{
- API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
- API_KEY_SYNC_GROUP, BrokerAdvertise, ERROR_GROUP_MAX_SIZE_REACHED, ERROR_ILLEGAL_GENERATION,
+ API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, API_KEY_JOIN_GROUP,
+ API_KEY_LEAVE_GROUP, API_KEY_LIST_GROUPS, API_KEY_SYNC_GROUP, BrokerAdvertise,
+ ERROR_GROUP_ID_NOT_FOUND, ERROR_GROUP_MAX_SIZE_REACHED, ERROR_ILLEGAL_GENERATION,
ERROR_INCONSISTENT_GROUP_PROTOCOL, ERROR_INVALID_GROUP_ID, ERROR_INVALID_REQUEST,
ERROR_INVALID_SESSION_TIMEOUT, ERROR_MEMBER_ID_REQUIRED, ERROR_NONE,
ERROR_REBALANCE_IN_PROGRESS, ERROR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED,
- ERROR_UNKNOWN_MEMBER_ID, GatewayState, handle_request_bounded,
+ ERROR_UNKNOWN_MEMBER_ID, ERROR_UNKNOWN_SERVER_ERROR, GatewayState, handle_request_bounded,
};
+use iggy_gateway_kafka::protocol::header::response_header_version;
use codec::Decoder;
use server::spawn_test_server_with_config;
use tcp::{build_request_frame, parse_response_payload, read_response_frame};
use wire::{
- JoinGroupParams, SyncGroupParams, build_find_coordinator_request, build_heartbeat_request,
- build_join_group_request, build_leave_group_request, build_sync_group_request,
+ JoinGroupParams, SyncGroupParams, build_describe_groups_request,
+ build_find_coordinator_request, build_heartbeat_request, build_join_group_request,
+ build_leave_group_request, build_list_groups_request, build_sync_group_request,
};
const GROUP: &str = "orders";
@@ -75,14 +78,27 @@ const REBALANCE_TIMEOUT_MS: i32 = 20_000;
// ── Fixtures ────────────────────────────────────────────────────────────────
fn test_state(config: GroupCoordinatorConfig) -> Arc<GatewayState> {
- Arc::new(GatewayState::new(
+ Arc::new(owned_state(config, 8 * 1024 * 1024))
+}
+
+fn owned_state(config: GroupCoordinatorConfig, max_frame_size: usize) -> GatewayState {
+ GatewayState::new(
BrokerAdvertise::default(),
None,
- 8 * 1024 * 1024,
+ max_frame_size,
false,
0,
GroupCoordinator::new(config, CancellationToken::new()),
- ))
+ )
+}
+
+fn response_frame_len(api_key: i16, version: i16, body: &Bytes) -> usize {
+ let header = if response_header_version(api_key, version) >= 1 {
+ 5
+ } else {
+ 4
+ };
+ header + body.len()
}
/// A coordinator that completes a new group's first join immediately, so tests that are not
@@ -1416,6 +1432,563 @@ async fn given_a_transaction_key_type_when_finding_the_coordinator_should_return
assert_eq!(decoder.remaining(), 0);
}
+// ── DescribeGroups / ListGroups ─────────────────────────────────────────────
+
+async fn describe(state: &GatewayState, version: i16, groups: &[&str]) -> DescribeResponse {
+ let body = build_describe_groups_request(version, groups, false);
+ let response = handle_request_bounded(state, API_KEY_DESCRIBE_GROUPS, version, body)
+ .await
+ .expect_response("DescribeGroups must answer");
+ DescribeResponse::decode(version, response)
+}
+
+async fn list_groups(
+ state: &GatewayState,
+ version: i16,
+ states: &[&str],
+ group_types: &[&str],
+) -> ListResponse {
+ let body = build_list_groups_request(version, states, group_types);
+ let response = handle_request_bounded(state, API_KEY_LIST_GROUPS, version, body)
+ .await
+ .expect_response("ListGroups must answer");
+ ListResponse::decode(version, response)
+}
+
+struct DescribeResponse {
+ groups: Vec<Described>,
+}
+
+struct Described {
+ error: i16,
+ error_message: Option<String>,
+ group_id: String,
+ state: String,
+ protocol_type: String,
+ protocol_data: String,
+ members: Vec<DescribedMember>,
+ authorized_operations: Option<i32>,
+}
+
+struct DescribedMember {
+ metadata: Bytes,
+ assignment: Bytes,
+}
+
+impl DescribeResponse {
+ fn decode(version: i16, body: Bytes) -> Self {
+ let flexible = version >= 5;
+ let mut decoder = Decoder::new(body);
+ if version >= 1 {
+ decoder.read_i32().unwrap();
+ }
+ let count = read_array_count(&mut decoder, flexible);
+ let mut groups = Vec::with_capacity(count);
+ for _ in 0..count {
+ let error = decoder.read_i16().unwrap();
+ let error_message = if version >= 6 {
+ decoder.read_compact_nullable_string().unwrap()
+ } else {
+ None
+ };
+ let group_id = read_nullable(&mut decoder, flexible).unwrap_or_default();
+ let state = read_nullable(&mut decoder, flexible).unwrap_or_default();
+ let protocol_type = read_nullable(&mut decoder, flexible).unwrap_or_default();
+ let protocol_data = read_nullable(&mut decoder, flexible).unwrap_or_default();
+ let member_count = read_array_count(&mut decoder, flexible);
+ let mut members = Vec::with_capacity(member_count);
+ for _ in 0..member_count {
+ let _member_id = read_nullable(&mut decoder, flexible);
+ if version >= 4 {
+ let _instance_id = read_nullable(&mut decoder, flexible);
+ }
+ let client_id = read_nullable(&mut decoder, flexible);
+ let client_host = read_nullable(&mut decoder, flexible);
+ assert_eq!(client_id.as_deref(), Some(""));
+ assert_eq!(client_host.as_deref(), Some(""));
+ let metadata = read_bytes_field(&mut decoder, flexible);
+ let assignment = read_bytes_field(&mut decoder, flexible);
+ if flexible {
+ decoder.read_tagged_fields().unwrap();
+ }
+ members.push(DescribedMember {
+ metadata,
+ assignment,
+ });
+ }
+ let authorized_operations = if version >= 3 {
+ Some(decoder.read_i32().unwrap())
+ } else {
+ None
+ };
+ if flexible {
+ decoder.read_tagged_fields().unwrap();
+ }
+ groups.push(Described {
+ error,
+ error_message,
+ group_id,
+ state,
+ protocol_type,
+ protocol_data,
+ members,
+ authorized_operations,
+ });
+ }
+ if flexible {
+ decoder.read_tagged_fields().unwrap();
+ }
+ assert_eq!(
+ decoder.remaining(),
+ 0,
+ "DescribeGroups v{version} trailing bytes"
+ );
+ Self { groups }
+ }
+}
+
+struct ListResponse {
+ error: i16,
+ groups: Vec<Listed>,
+}
+
+struct Listed {
+ group_id: String,
+ protocol_type: String,
+ state: Option<String>,
+ group_type: Option<String>,
+}
+
+impl ListResponse {
+ fn decode(version: i16, body: Bytes) -> Self {
+ let flexible = version >= 3;
+ let mut decoder = Decoder::new(body);
+ if version >= 1 {
+ decoder.read_i32().unwrap();
+ }
+ let error = decoder.read_i16().unwrap();
+ let count = read_array_count(&mut decoder, flexible);
+ let mut groups = Vec::with_capacity(count);
+ for _ in 0..count {
+ let group_id = read_nullable(&mut decoder, flexible).unwrap_or_default();
+ let protocol_type = read_nullable(&mut decoder, flexible).unwrap_or_default();
+ let state = if version >= 4 {
+ read_nullable(&mut decoder, flexible)
+ } else {
+ None
+ };
+ let group_type = if version >= 5 {
+ read_nullable(&mut decoder, flexible)
+ } else {
+ None
+ };
+ if flexible {
+ decoder.read_tagged_fields().unwrap();
+ }
+ groups.push(Listed {
+ group_id,
+ protocol_type,
+ state,
+ group_type,
+ });
+ }
+ if flexible {
+ decoder.read_tagged_fields().unwrap();
+ }
+ assert_eq!(
+ decoder.remaining(),
+ 0,
+ "ListGroups v{version} trailing bytes"
+ );
+ Self { error, groups }
+ }
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_running_group_when_describing_and_listing_should_report_members_assignment_and_state()
+ {
+ let state = test_state(immediate_config());
+ let (leader, follower) = two_member_group(&state).await;
+
+ let completing = describe(&state, 6, &[GROUP]).await;
+ assert_eq!(completing.groups.len(), 1);
+ let group = &completing.groups[0];
+ assert_eq!(group.error, ERROR_NONE);
+ assert_eq!(group.group_id, GROUP);
+ assert_eq!(group.state, "CompletingRebalance");
+ assert_eq!(group.protocol_type, "consumer");
+ assert_eq!(group.protocol_data, "");
+ assert_eq!(group.authorized_operations, Some(i32::MIN));
+ assert_eq!(group.members.len(), 2);
+ assert!(
+ group
+ .members
+ .iter()
+ .all(|member| member.metadata.is_empty()),
+ "metadata is the previous generation until the group is Stable"
+ );
+ assert!(
+ group
+ .members
+ .iter()
+ .all(|member| member.assignment.is_empty())
+ );
+
+ let listed = list_groups(&state, 5, &[], &[]).await;
+ assert_eq!(listed.error, ERROR_NONE);
+ assert_eq!(listed.groups.len(), 1);
+ assert_eq!(listed.groups[0].group_id, GROUP);
+ assert_eq!(listed.groups[0].protocol_type, "consumer");
+ assert_eq!(
+ listed.groups[0].state.as_deref(),
+ Some("CompletingRebalance")
+ );
+ assert_eq!(listed.groups[0].group_type.as_deref(), Some("classic"));
+
+ let leader_blob: &[u8] = b"partitions-0-1";
+ let follower_blob: &[u8] = b"partitions-2-3";
+ let synced = sync(
+ &state,
+ SYNC_VERSION,
+ &SyncGroupParams {
+ group_id: GROUP,
+ generation_id: 2,
+ member_id: &leader,
+ assignments: &[
+ (leader.as_str(), leader_blob),
+ (follower.as_str(), follower_blob),
+ ],
+ ..SyncGroupParams::default()
+ },
+ )
+ .await;
+ assert_eq!(synced.error, ERROR_NONE);
+
+ let stable = describe(&state, 0, &[GROUP]).await;
+ let group = &stable.groups[0];
+ assert_eq!(group.error, ERROR_NONE);
+ assert_eq!(group.state, "Stable");
+ assert_eq!(group.protocol_data, "range");
+ assert!(group.authorized_operations.is_none());
+ assert!(
+ group
+ .members
+ .iter()
+ .any(|member| member.metadata.as_ref() == b"leader-subscription")
+ );
+ assert!(
+ group
+ .members
+ .iter()
+ .any(|member| member.assignment.as_ref() == leader_blob)
+ );
+ assert!(
+ group
+ .members
+ .iter()
+ .any(|member| member.assignment.as_ref() == follower_blob)
+ );
+
+ let listed = list_groups(&state, 0, &[], &[]).await;
+ assert_eq!(listed.error, ERROR_NONE);
+ assert_eq!(listed.groups.len(), 1);
+ assert_eq!(listed.groups[0].group_id, GROUP);
+ assert_eq!(listed.groups[0].protocol_type, "consumer");
+ assert_eq!(listed.groups[0].state, None);
+
+ let stable_only = list_groups(&state, 5, &["stable"], &[]).await;
+ assert_eq!(stable_only.groups.len(), 1);
+ let preparing_only = list_groups(&state, 5, &["PreparingRebalance"], &[]).await;
+ assert!(preparing_only.groups.is_empty());
+ let unknown_state = list_groups(&state, 5, &["not-a-state"], &[]).await;
+ assert!(unknown_state.groups.is_empty());
+ let classic = list_groups(&state, 5, &[], &["CLASSIC"]).await;
+ assert_eq!(classic.groups.len(), 1);
+ let consumer_type = list_groups(&state, 5, &[], &["consumer"]).await;
+ assert!(consumer_type.groups.is_empty());
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_joining_member_when_describing_should_report_preparing_rebalance() {
+ let state = test_state(GroupCoordinatorConfig::default());
+ let leader = claim_member_id(&state, b"sub").await;
+ let parked = {
+ let state = Arc::clone(&state);
+ let leader = leader.clone();
+ tokio::spawn(async move {
+ let protocols: &[(&str, &[u8])] = &[("range", b"sub")];
+ join(&state, JOIN_VERSION, &join_params(&leader, protocols)).await
+ })
+ };
+ yield_to_parked().await;
+
+ let described = describe(&state, 6, &[GROUP]).await;
+ assert_eq!(described.groups[0].state, "PreparingRebalance");
+ assert_eq!(described.groups[0].protocol_data, "");
+ assert_eq!(described.groups[0].members.len(), 1);
+ assert!(described.groups[0].members[0].metadata.is_empty());
+ assert!(described.groups[0].members[0].assignment.is_empty());
+
+ let listed = list_groups(&state, 4, &["preparingrebalance"], &[]).await;
+ assert_eq!(listed.groups.len(), 1);
+ assert_eq!(
+ listed.groups[0].state.as_deref(),
+ Some("PreparingRebalance")
+ );
+
+ advance(Duration::from_secs(3)).await;
+ let joined = parked.await.expect("parked JoinGroup task");
+ assert_eq!(joined.error, ERROR_NONE);
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_only_a_pending_member_when_listing_should_report_empty() {
+ let state = test_state(immediate_config());
+ let _member = claim_member_id(&state, b"sub").await;
+
+ let described = describe(&state, 6, &[GROUP]).await;
+ assert_eq!(described.groups[0].error, ERROR_NONE);
+ assert_eq!(described.groups[0].state, "Empty");
+ assert!(described.groups[0].members.is_empty());
+
+ let empty_only = list_groups(&state, 5, &["Empty"], &[]).await;
+ assert_eq!(empty_only.groups.len(), 1);
+ assert_eq!(empty_only.groups[0].state.as_deref(), Some("Empty"));
+ let preparing = list_groups(&state, 5, &["PreparingRebalance"], &[]).await;
+ assert!(preparing.groups.is_empty());
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_an_expired_group_when_listing_should_omit_it_and_describing_should_report_dead() {
+ let state = test_state(immediate_config());
+ let (leader, _follower) = two_member_group(&state).await;
+ let synced = sync(
+ &state,
+ SYNC_VERSION,
+ &SyncGroupParams {
+ group_id: GROUP,
+ generation_id: 2,
+ member_id: &leader,
+ assignments: &[(&leader, b"partitions-0")],
+ ..SyncGroupParams::default()
+ },
+ )
+ .await;
+ assert_eq!(synced.error, ERROR_NONE);
+
+ advance(Duration::from_secs(11)).await;
+ let listed = list_groups(&state, 5, &[], &[]).await;
+ assert!(
+ listed.groups.is_empty(),
+ "a group whose sessions are all past is omitted, and listing does not have to tick"
+ );
+
+ let described = describe(&state, 6, &[GROUP]).await;
+ assert_eq!(described.groups[0].state, "Dead");
+ assert_eq!(described.groups[0].error, ERROR_GROUP_ID_NOT_FOUND);
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_an_unknown_group_when_describing_should_report_dead_without_creating_it() {
+ let state = test_state(immediate_config());
+
+ let v6 = describe(&state, 6, &["missing"]).await;
+ assert_eq!(v6.groups[0].error, ERROR_GROUP_ID_NOT_FOUND);
+ assert_eq!(
+ v6.groups[0].error_message.as_deref(),
+ Some("Group missing not found.")
+ );
+ assert_eq!(v6.groups[0].state, "Dead");
+ assert!(v6.groups[0].members.is_empty());
+ assert_eq!(v6.groups[0].authorized_operations, Some(i32::MIN));
+
+ let v0 = describe(&state, 0, &["missing"]).await;
+ assert_eq!(v0.groups[0].error, ERROR_NONE);
+ assert_eq!(v0.groups[0].error_message, None);
+ assert_eq!(v0.groups[0].state, "Dead");
+ assert!(v0.groups[0].authorized_operations.is_none());
+
+ let with_ops = handle_request_bounded(
+ &state,
+ API_KEY_DESCRIBE_GROUPS,
+ 3,
+ build_describe_groups_request(3, &["missing"], true),
+ )
+ .await
+ .expect_response("DescribeGroups must answer");
+ let decoded = DescribeResponse::decode(3, with_ops);
+ assert_eq!(decoded.groups[0].authorized_operations, Some(i32::MIN));
+
+ let listed = list_groups(&state, 0, &[], &[]).await;
+ assert_eq!(listed.error, ERROR_NONE);
+ assert!(listed.groups.is_empty());
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_repeated_group_ids_when_describing_should_answer_each_distinct_id_once() {
+ let state = test_state(immediate_config());
+ let protocols: &[(&str, &[u8])] = &[("range", b"sub")];
+ join(&state, 3, &join_params("", protocols)).await;
+
+ let described = describe(&state, 6, &[GROUP, "missing", GROUP, "missing"]).await;
+
+ assert_eq!(described.groups.len(), 2);
+ assert_eq!(described.groups[0].group_id, GROUP);
+ assert_eq!(described.groups[0].error, ERROR_NONE);
+ assert_eq!(described.groups[1].group_id, "missing");
+ assert_eq!(described.groups[1].state, "Dead");
+ assert_eq!(described.groups[1].error, ERROR_GROUP_ID_NOT_FOUND);
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_describe_response_one_byte_over_the_frame_when_describing_should_summarize() {
+ let mut state = owned_state(immediate_config(), 8 * 1024 * 1024);
+ let blob = vec![b'x'; 400];
+ let protocols: &[(&str, &[u8])] = &[("range", blob.as_slice())];
+ let joined = join(&state, 3, &join_params("", protocols)).await;
+ assert_eq!(joined.error, ERROR_NONE);
+ let assignment = vec![b'y'; 400];
+ let synced = sync(
+ &state,
+ 3,
+ &SyncGroupParams {
+ group_id: GROUP,
+ generation_id: joined.generation_id,
+ member_id: &joined.member_id,
+ assignments: &[(&joined.member_id, assignment.as_slice())],
+ ..SyncGroupParams::default()
+ },
+ )
+ .await;
+ assert_eq!(synced.error, ERROR_NONE);
+
+ let version = 0;
+ let encoded = handle_request_bounded(
+ &state,
+ API_KEY_DESCRIBE_GROUPS,
+ version,
+ build_describe_groups_request(version, &[GROUP], false),
+ )
+ .await
+ .expect_response("DescribeGroups must answer");
+ let frame = response_frame_len(API_KEY_DESCRIBE_GROUPS, version, &encoded);
+ state.max_frame_size = frame;
+ let repeated = describe(&state, version, &[GROUP, GROUP]).await;
+ assert_eq!(
+ repeated.groups.len(),
+ 1,
+ "a repeated id is one group, so it still fits the one-group frame"
+ );
+ assert!(!repeated.groups[0].members[0].assignment.is_empty());
+
+ state.max_frame_size = frame - 1;
+ let summarized = describe(&state, version, &[GROUP]).await;
+ assert_eq!(summarized.groups[0].error, ERROR_NONE);
+ assert_eq!(summarized.groups[0].state, "Stable");
+ assert_eq!(summarized.groups[0].protocol_data, "");
+ assert!(summarized.groups[0].members[0].metadata.is_empty());
+ assert!(summarized.groups[0].members[0].assignment.is_empty());
+
+ let summary_body = handle_request_bounded(
+ &state,
+ API_KEY_DESCRIBE_GROUPS,
+ version,
+ build_describe_groups_request(version, &[GROUP], false),
+ )
+ .await
+ .expect_response("a summarized DescribeGroups must answer");
+ let summary_frame = response_frame_len(API_KEY_DESCRIBE_GROUPS, version, &summary_body);
+ state.max_frame_size = summary_frame - 1;
+ let errored = describe(&state, version, &[GROUP]).await;
+ assert_eq!(errored.groups[0].error, ERROR_UNKNOWN_SERVER_ERROR);
+ assert!(errored.groups[0].members.is_empty());
+
+ state.max_frame_size = frame;
+ let again = describe(&state, version, &[GROUP]).await;
+ assert_eq!(again.groups.len(), 1);
+ assert_eq!(again.groups[0].group_id, GROUP);
+ assert_eq!(again.groups[0].state, "Stable");
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_list_response_one_byte_past_the_frame_cap_when_listing_should_truncate() {
+ let mut state = owned_state(immediate_config(), 8 * 1024 * 1024);
+ let protocols: &[(&str, &[u8])] = &[("range", b"sub")];
+ join(&state, 3, &join_params("", protocols)).await;
+
+ let version = 0;
+ let encoded = handle_request_bounded(
+ &state,
+ API_KEY_LIST_GROUPS,
+ version,
+ build_list_groups_request(version, &[], &[]),
+ )
+ .await
+ .expect_response("ListGroups must answer");
+ let frame = response_frame_len(API_KEY_LIST_GROUPS, version, &encoded);
+
+ // One byte under what the one listed group needs: truncated to zero groups rather than
+ // closing the connection, since this is server-sized state outgrowing a client-chosen
+ // frame size, not a malformed request.
+ state.max_frame_size = frame - 1;
+ let truncated = list_groups(&state, version, &[], &[]).await;
+ assert!(
+ truncated.groups.is_empty(),
+ "ListGroups truncates rather than closing when the encoded frame would pass \
+ max_frame_size"
+ );
+
+ state.max_frame_size = frame;
+ let listed = list_groups(&state, version, &[], &[]).await;
+ assert_eq!(listed.groups.len(), 1);
+ assert_eq!(listed.groups[0].group_id, GROUP);
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_truncated_describe_groups_body_when_handled_should_close_the_connection() {
+ let state = test_state(immediate_config());
+ let outcome = handle_request_bounded(
+ &state,
+ API_KEY_DESCRIBE_GROUPS,
+ 0,
+ Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x01, // one group
+ 0x00, 0x02, // string length 2
+ b'g', // truncated
+ ]),
+ )
+ .await;
+
+ assert!(
+ outcome.is_close(),
+ "a truncated DescribeGroups body must close the connection"
+ );
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_truncated_list_groups_body_when_handled_should_return_invalid_request() {
+ let state = test_state(immediate_config());
+ let response = handle_request_bounded(
+ &state,
+ API_KEY_LIST_GROUPS,
+ 4,
+ Bytes::from_static(&[
+ 0x02, // one states_filter entry
+ 0x03, // compact string claims 2 bytes
+ b'x', // truncated
+ ]),
+ )
+ .await
+ .expect_response("ListGroups reports a decode failure as INVALID_REQUEST");
+ let mut decoder = Decoder::new(response);
+
+ assert_eq!(decoder.read_i32().unwrap(), 0);
+ assert_eq!(decoder.read_i16().unwrap(), ERROR_INVALID_REQUEST);
+ assert_eq!(decoder.read_varint().unwrap(), 1);
+ assert_eq!(decoder.read_varint().unwrap(), 0);
+ assert_eq!(decoder.remaining(), 0);
+}
+
// ── Over a real TCP listener ────────────────────────────────────────────────
/// The in-process tests drive one `GatewayState` directly. This one proves the connection loop
@@ -1599,6 +2172,78 @@ async fn tcp_leave_scenario(leave_version: i16) {
assert_eq!(alone.members.len(), 1, "v{leave_version}");
}
+/// The `given_a_running_group_when_describing_and_listing_*` tests above all drive
+/// `handle_request_bounded` directly, bypassing real TCP dispatch and response headers. This
+/// creates a live group through a real `JoinGroup`/`SyncGroup` round trip over one socket, then
+/// issues real `DescribeGroups` and `ListGroups` requests on that same connection and decodes
+/// their actual wire responses - the real-wire harness this crate already uses for `JoinGroup`/
+/// `SyncGroup`/`LeaveGroup` above, now covering the admin-view APIs those tests never reached.
+#[tokio::test]
+async fn given_a_live_group_over_tcp_should_describe_and_list_it_through_the_real_listener() {
+ let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig {
+ group: immediate_config(),
+ ..GatewayConfig::default()
+ })
+ .await;
+ let mut stream = TcpStream::connect(addr).await.expect("connect");
+
+ let protocols: &[(&str, &[u8])] = &[("range", b"sub")];
+ // KIP-394: the first join with no member id only allocates one and answers
+ // MEMBER_ID_REQUIRED; the member must rejoin with that id to actually complete the join.
+ let member = tcp_join(&mut stream, &join_params("", protocols))
+ .await
+ .member_id;
+ let joined = tcp_join(&mut stream, &join_params(&member, protocols)).await;
+ assert_eq!(joined.generation_id, 1);
+ write_request(
+ &mut stream,
+ API_KEY_SYNC_GROUP,
+ SYNC_VERSION,
+ 2,
+ &build_sync_group_request(
+ SYNC_VERSION,
+ &SyncGroupParams {
+ group_id: GROUP,
+ generation_id: 1,
+ member_id: &member,
+ assignments: &[(member.as_str(), b"partitions-0")],
+ ..SyncGroupParams::default()
+ },
+ ),
+ )
+ .await;
+ let synced = read_sync(&mut stream).await;
+ assert_eq!(synced.assignment.as_ref(), b"partitions-0");
+
+ write_request(
+ &mut stream,
+ API_KEY_DESCRIBE_GROUPS,
+ 6,
+ 3,
+ &build_describe_groups_request(6, &[GROUP], false),
+ )
+ .await;
+ let described = read_describe_groups(&mut stream, 6).await;
+ assert_eq!(described.groups.len(), 1);
+ assert_eq!(described.groups[0].group_id, GROUP);
+ assert_eq!(described.groups[0].state, "Stable");
+ assert_eq!(described.groups[0].members.len(), 1);
+
+ write_request(
+ &mut stream,
+ API_KEY_LIST_GROUPS,
+ 5,
+ 4,
+ &build_list_groups_request(5, &[], &[]),
+ )
+ .await;
+ let listed = read_list_groups(&mut stream, 5).await;
+ assert!(
+ listed.groups.iter().any(|group| group.group_id == GROUP),
+ "the real-TCP DescribeGroups/SyncGroup group must also be visible to ListGroups"
+ );
+}
+
async fn write_request(
stream: &mut TcpStream,
api_key: i16,
@@ -1639,3 +2284,15 @@ async fn read_sync(stream: &mut TcpStream) -> SyncResponse {
let (_, body) = parse_response_payload(API_KEY_SYNC_GROUP, SYNC_VERSION, payload);
SyncResponse::decode(SYNC_VERSION, body)
}
+
+async fn read_describe_groups(stream: &mut TcpStream, version: i16) -> DescribeResponse {
+ let payload = read_response_frame(stream, 8 * 1024 * 1024).await;
+ let (_, body) = parse_response_payload(API_KEY_DESCRIBE_GROUPS, version, payload);
+ DescribeResponse::decode(version, body)
+}
+
+async fn read_list_groups(stream: &mut TcpStream, version: i16) -> ListResponse {
+ let payload = read_response_frame(stream, 8 * 1024 * 1024).await;
+ let (_, body) = parse_response_payload(API_KEY_LIST_GROUPS, version, payload);
+ ListResponse::decode(version, body)
+}
diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
index d1e763690..e5cc16210 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=17 (compact array: N+1=18)
+ // error_code=0, api_count=19 (compact array: N+1=20)
// each entry followed by an empty tagged-fields byte; throttle_ms=0; top-level tagged fields
- let expected: [u8; 127] = [
+ let expected: [u8; 141] = [
0x00, 0x00, // error_code
- 0x12, // compact array count (17+1)
+ 0x14, // compact array count (19+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
@@ -57,6 +57,8 @@ async fn golden_apiversions_v3_flexible_response_fixture() {
0x00, 0x0C, 0x00, 0x00, 0x00, 0x04, 0x00, // key 12: Heartbeat 0-4
0x00, 0x0D, 0x00, 0x00, 0x00, 0x05, 0x00, // key 13: LeaveGroup 0-5
0x00, 0x0E, 0x00, 0x00, 0x00, 0x05, 0x00, // key 14: SyncGroup 0-5
+ 0x00, 0x0F, 0x00, 0x00, 0x00, 0x06, 0x00, // key 15: DescribeGroups 0-6
+ 0x00, 0x10, 0x00, 0x00, 0x00, 0x05, 0x00, // key 16: ListGroups 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
@@ -76,10 +78,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=17; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
- let expected: [u8; 112] = [
+ // error_code=0, api_count=19; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
+ let expected: [u8; 124] = [
0x00, 0x00, // error_code
- 0x00, 0x00, 0x00, 0x11, // api count = 17
+ 0x00, 0x00, 0x00, 0x13, // api count = 19
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
@@ -91,6 +93,8 @@ async fn golden_apiversions_v1_response_fixture() {
0x00, 0x0C, 0x00, 0x00, 0x00, 0x04, // key 12: Heartbeat 0-4
0x00, 0x0D, 0x00, 0x00, 0x00, 0x05, // key 13: LeaveGroup 0-5
0x00, 0x0E, 0x00, 0x00, 0x00, 0x05, // key 14: SyncGroup 0-5
+ 0x00, 0x0F, 0x00, 0x00, 0x00, 0x06, // key 15: DescribeGroups 0-6
+ 0x00, 0x10, 0x00, 0x00, 0x00, 0x05, // key 16: ListGroups 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
diff --git a/gateways/kafka/tests/server_e2e_tests.rs b/gateways/kafka/tests/server_e2e_tests.rs
index d9105a6a4..d8fa162db 100644
--- a/gateways/kafka/tests/server_e2e_tests.rs
+++ b/gateways/kafka/tests/server_e2e_tests.rs
@@ -105,9 +105,9 @@ async fn e2e_unsupported_api_key_closes_connection() {
let (addr, _shutdown) = spawn_test_server().await;
let mut stream = TcpStream::connect(addr).await.unwrap();
- // Unknown api key (16, ListGroups) has no response schema this gateway can encode, so the
+ // Unknown api key (21, DeleteRecords) has no response schema this gateway can encode, so the
// server closes the connection without a (misparseable) response body.
- let frame1 = build_request_frame(16, 2, 99, Some("e2e-test"), &[]);
+ let frame1 = build_request_frame(21, 2, 99, Some("e2e-test"), &[]);
stream.write_all(&frame1).await.unwrap();
assert_eq!(
diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs
index c3b7620f3..e0dce7872 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -38,11 +38,12 @@ use tokio::net::TcpStream;
use iggy_gateway_kafka::protocol::api::{
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,
+ API_KEY_DESCRIBE_CONFIGS, API_KEY_DESCRIBE_GROUPS, 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_GROUPS, 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;
@@ -58,16 +59,17 @@ 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_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_describe_groups_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_groups_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_seventeen_entries() {
- assert_eq!(supported_api_ranges().len(), 17);
+fn supported_ranges_table_has_nineteen_entries() {
+ assert_eq!(supported_api_ranges().len(), 19);
}
#[test]
@@ -105,9 +107,9 @@ 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 (Produce, Fetch, `ListOffsets`, Metadata, `OffsetCommit`, `OffsetFetch`,
-/// `FindCoordinator`, `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`, `ApiVersions`,
-/// `CreateTopics`, `DeleteTopics`, `InitProducerId`, `DescribeConfigs`, `AlterConfigs`).
-/// `supported_ranges_table_has_seventeen_entries` and
+/// `FindCoordinator`, `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`, `DescribeGroups`,
+/// `ListGroups`, `ApiVersions`, `CreateTopics`, `DeleteTopics`, `InitProducerId`,
+/// `DescribeConfigs`, `AlterConfigs`). `supported_ranges_table_has_nineteen_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() {
@@ -355,7 +357,7 @@ async fn create_topics_below_min_version_closes_connection() {
#[tokio::test]
async fn unsupported_api_keys_close_connection() {
- for key in [8, 9, 15, 16, 20, 42, 999] {
+ for key in [8, 9, 20, 42, 999] {
let outcome = handle_request(key, 0, Bytes::new(), &default_broker()).await;
assert!(
outcome.is_close(),
@@ -525,6 +527,8 @@ fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes
},
),
API_KEY_INIT_PRODUCER_ID => build_init_producer_id_request(version, None),
+ API_KEY_DESCRIBE_GROUPS => build_describe_groups_request(version, &["scope-group"], false),
+ API_KEY_LIST_GROUPS => build_list_groups_request(version, &[], &[]),
API_KEY_DESCRIBE_CONFIGS => build_describe_configs_empty_request(version),
API_KEY_ALTER_CONFIGS => build_alter_configs_empty_request(version),
API_KEY_OFFSET_COMMIT => build_offset_commit_request(version, "scope-group", "orders", 0),