You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ra...@apache.org on 2026/10/05 00:38:09 UTC
(iggy) branch feat/gateways-3548-describe-list-groups created (now 356839c02)
This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a change to branch feat/gateways-3548-describe-list-groups
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 356839c02 Intial version of Describe and List Groups
This branch includes the following new commits:
new 85e45e13e Intial version of Describe and List Groups
new 356839c02 Intial version of Describe and List Groups
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/02: Intial version of Describe and List Groups
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch feat/gateways-3548-describe-list-groups
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 85e45e13e3f7a7570a033537d9e558818718254f
Author: ryerraguntla <ra...@gmail.com>
AuthorDate: Sun Oct 4 20:34:39 2026 -0400
Intial version of Describe and List Groups
---
gateways/kafka/README.md | 5 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 21 +-
gateways/kafka/docs/MANUAL_TESTING.md | 6 +-
gateways/kafka/docs/SCOPE.md | 12 +-
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/src/group/mod.rs | 61 +++-
gateways/kafka/src/group/state.rs | 75 ++++-
gateways/kafka/src/protocol/api.rs | 16 +-
gateways/kafka/src/protocol/bounds_guard.rs | 123 ++++++-
.../kafka/src/protocol/handlers/describe_groups.rs | 142 +++++++++
.../kafka/src/protocol/handlers/list_groups.rs | 114 +++++++
gateways/kafka/src/protocol/handlers/mod.rs | 15 +-
gateways/kafka/tests/common/scope.rs | 2 +
gateways/kafka/tests/common/wire.rs | 46 ++-
gateways/kafka/tests/consumer_group_tests.rs | 355 ++++++++++++++++++++-
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 16 +-
gateways/kafka/tests/server_e2e_tests.rs | 2 +-
gateways/kafka/tests/version_firewall_tests.rs | 31 +-
18 files changed, 990 insertions(+), 54 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index e8841b8d6..c19990cb8 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -13,8 +13,9 @@ 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)). With the bridge off,
> Metadata reports every topic unknown, so a consumer joins a group and is assigned 0 partitions.
> Offset commit and fetch are not implemented yet, so a consumer must use `assign()` with explicit
> start offsets and `enable.auto.commit=false`. The Java client needs no `group.id`. librdkafka
diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md
index f649147e9..7addf9b56 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,28 @@ 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 `PreparingRebalance`, `CompletingRebalance` and `Stable`. `Empty` is not stored: a group with
+no members is removed. `Dead` is only the describe answer for an id that is not in the map.
+
+`ListGroups` does not expire sessions. Expiry runs when a request names a group, and a list names
+none, so a quiet group stays listed until something else touches it. `DescribeGroups` does tick
+each named group, so it reports the membership a heartbeat would see.
+
+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
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 3df52b832..a01faa9b4 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -221,14 +221,16 @@ 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 |
| ID | Test | Expected for in-range | Expected for out-of-range |
| ---- | ------ | ---------------------- | --------------------------- |
-| B1 | ApiVersions negotiation | `error_code=0`; body lists 12 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 |
+| B1 | ApiVersions negotiation | `error_code=0`; body lists 14 API keys with correct min/max | KIP-511 exception: still answers, `error_code=35` (UNSUPPORTED_VERSION), v0 response header regardless of the request's own encoding |
| B2 | Metadata out-of-range | N/A | **Connection closes**, no response sent - Metadata has no top-level error field to carry a version-correct error in |
| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these five messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
-| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,10,11,12,13,14,18,19,22 only — no OffsetCommit/OffsetFetch, and no transaction keys (24, 25, 26, 28) |
+| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,10,11,12,13,14,15,16,18,19,22 only. No OffsetCommit/OffsetFetch, 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
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index b2c736040..ecc996ebd 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -29,7 +29,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, and 0 for the four consumer-group keys), so there is no version below an API's
+CreateTopics 2, InitProducerId 0, and 0 for the consumer-group keys), 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.
@@ -54,6 +54,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+ |
A request is accepted when `min_version ≤ api_version ≤ max_version` for that API key. Any other version for a listed key closes the connection (ApiVersions excepted - see Governance model above).
@@ -75,6 +77,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 |
| 22 | InitProducerId | 0–5 | v2 |
@@ -89,7 +93,6 @@ All API keys not listed above close the connection (see Governance model above)
| --------- | ------ | ------- |
| 8 | OffsetCommit | Consumer group offsets — [#3542](https://github.com/apache/iggy/issues/3542) |
| 9 | OffsetFetch | Consumer group offsets — [#3542](https://github.com/apache/iggy/issues/3542); sent right after SyncGroup, so a joined consumer loops on it today ([`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)) |
-| 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)) |
@@ -157,7 +160,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 12 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 and CreateTopics with a bridge |
+| **2 — Request/response codecs** | Partial | Decode/encode via the `kafka_protocol` crate (broker feature only) for 14 keys; `bounds_guard.rs` pre-validates against unbounded allocation before handing a frame to the crate; stub responses except InitProducerId and the consumer-group keys, and Produce, Fetch, ListOffsets, Metadata and CreateTopics with a bridge |
| **3 — Iggy bridge** | Produce, Fetch, ListOffsets, Metadata and CreateTopics wired | `bridge/` module (connection, topic mapping, provisioning, high watermark, `topic_target` + `send_records`, `probe` + `poll`). Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch ([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets ([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)) and CreateTopics ([#3538 [...]
---
@@ -348,7 +351,8 @@ Offset persistence design ([#3540](https://github.com/apache/iggy/issues/3540)):
[#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)
- [ ] OffsetCommit (8), OffsetFetch (9)
-- [ ] DescribeGroups (15), ListGroups (16) as needed by target clients
+- [x] DescribeGroups (15), ListGroups (16) -
+ [#3548](https://github.com/apache/iggy/issues/3548); see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)
### Phase 3+ — Auth, admin, tuning
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index e2a006958..301e7ced5 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/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index 07cd1a891..b445d2120 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -18,9 +18,9 @@
//! In-memory coordinator for Kafka's classic consumer group protocol.
//!
//! [`GroupCoordinator`] owns every group this gateway instance coordinates and is the only
-//! module that awaits: `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup` handlers
-//! translate wire messages into the request types here, and `state` holds the synchronous state
-//! machine those requests drive.
+//! module that awaits: `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup`/
+//! `DescribeGroups`/`ListGroups` 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`.
@@ -40,6 +40,13 @@ 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 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";
+
/// 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.
@@ -320,6 +327,36 @@ 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,
+ /// One of the `GROUP_STATE_*` strings.
+ pub state: &'static str,
+ pub protocol_type: StrBytes,
+ /// Selected protocol name. `None` until the first join barrier completes.
+ pub protocol_name: Option<StrBytes>,
+ pub members: Vec<MemberDescription>,
+}
+
/// The answer for one `LeavingMember`.
#[derive(Debug, Clone)]
pub struct LeftMember {
@@ -451,6 +488,24 @@ impl GroupCoordinator {
state::leave_step(&mut groups, request, Instant::now())
}
+ /// Snapshot of each named group, in request order. `None` means the group is not here.
+ ///
+ /// Ticks each named group first, the same way every other request for that group does, so the
+ /// snapshot is the membership a heartbeat arriving now would see.
+ pub async fn describe_groups(&self, group_ids: &[StrBytes]) -> Vec<Option<GroupDescription>> {
+ let mut groups = self.groups.lock().await;
+ state::describe_groups(&mut groups, group_ids, Instant::now())
+ }
+
+ /// Every group currently in the map, ordered by group id.
+ ///
+ /// Does not expire sessions. Expiry runs on the request that names a group; a list names
+ /// none, and sweeping every group here would open rebalances as a side effect of reading.
+ pub async fn list_groups(&self) -> Vec<GroupListing> {
+ let groups = self.groups.lock().await;
+ state::list_groups(&groups)
+ }
+
/// 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 {
tokio::select! {
diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs
index 1597fb922..8bfcb9b89 100644
--- a/gateways/kafka/src/group/state.rs
+++ b/gateways/kafka/src/group/state.rs
@@ -35,8 +35,10 @@ use tokio::time::Instant;
use uuid::Uuid;
use crate::group::{
- GroupCoordinatorConfig, JoinRequest, JoinResult, JoinedMember, LeaveRequest, LeaveResult,
- LeavingMember, LeftMember, SyncRequest, SyncResult, owned_str,
+ GROUP_STATE_COMPLETING_REBALANCE, GROUP_STATE_PREPARING_REBALANCE, GROUP_STATE_STABLE,
+ GroupCoordinatorConfig, GroupDescription, GroupListing, 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,
@@ -71,6 +73,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),
@@ -619,6 +631,29 @@ impl GroupState {
self.sync_deadline = None;
self.bump();
}
+
+ fn description(&self, group_id: &StrBytes) -> GroupDescription {
+ let protocol_name = 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: protocol_name
+ .as_ref()
+ .map_or_else(Bytes::new, |name| member.metadata_for(name)),
+ assignment: member.assignment.clone(),
+ })
+ .collect();
+ GroupDescription {
+ group_id: group_id.clone(),
+ state: self.phase.as_kafka_state(),
+ protocol_type: self.protocol_type.clone(),
+ protocol_name,
+ members,
+ }
+ }
}
/// Tick every group and drop the ones that emptied, returning how many were reclaimed.
@@ -1112,6 +1147,42 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// One entry per `group_ids` element, after that group's usual expiry tick.
+pub fn describe_groups(
+ groups: &mut Groups,
+ group_ids: &[StrBytes],
+ now: Instant,
+) -> Vec<Option<GroupDescription>> {
+ let mut described = Vec::with_capacity(group_ids.len());
+ for group_id in group_ids {
+ if !tick_group(groups, group_id, now) {
+ described.push(None);
+ continue;
+ }
+ let Some(group) = groups.get(group_id) else {
+ described.push(None);
+ continue;
+ };
+ described.push(Some(group.description(group_id)));
+ }
+ described
+}
+
+/// Groups currently in the map, ordered by group id. Does not tick: see
+/// [`crate::group::GroupCoordinator::list_groups`].
+pub fn list_groups(groups: &Groups) -> Vec<GroupListing> {
+ let mut listed: Vec<GroupListing> = groups
+ .iter()
+ .map(|(group_id, group)| GroupListing {
+ group_id: group_id.clone(),
+ protocol_type: group.protocol_type.clone(),
+ state: group.phase.as_kafka_state(),
+ })
+ .collect();
+ listed.sort_by(|left, right| left.group_id.cmp(&right.group_id));
+ listed
+}
+
fn admit(
group: &mut GroupState,
config: &GroupCoordinatorConfig,
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index c2478ecbf..9aab821c9 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -38,9 +38,9 @@ use crate::protocol::bounds_guard::{
};
use crate::protocol::handlers::init_producer_id::ProducerIdAllocator;
use crate::protocol::handlers::{
- api_versions, create_topics, decode_guarded, dispatch, fetch, find_coordinator, heartbeat,
- init_producer_id, join_group, leave_group, list_offsets, metadata, produce, respond_or_close,
- sync_group,
+ api_versions, create_topics, decode_guarded, describe_groups, dispatch, fetch,
+ find_coordinator, heartbeat, init_producer_id, join_group, leave_group, list_groups,
+ list_offsets, metadata, produce, respond_or_close, sync_group,
};
use crate::protocol::probe_board::ProbeBoard;
use crate::protocol::sasl::{
@@ -56,6 +56,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;
@@ -103,6 +105,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;
@@ -278,6 +286,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 ae265b8ed..2e6640094 100644
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@ -26,8 +26,8 @@
//! topics count requests ~481 GB; a 4-byte Metadata v0 frame requests ~143 GiB. No SASL/TLS gates
//! any of `SUPPORTED_RANGES`, so this is reachable by anyone who can `connect()`.
//!
-//! This module walks the same field shape `kafka_protocol`'s real decode walks for each of the
-//! eleven accepted message types, but only to validate every length-prefixed field (array count,
+//! This module walks the same field shape `kafka_protocol`'s real decode walks for each accepted
+//! message type, but only to validate every length-prefixed field (array count,
//! string length, bytes length, tagged-field size) against what could still fit in the bytes
//! remaining in the frame - it never materializes a value or allocates a collection. Call the
//! matching `validate_*_shape` function before handing the body to `kafka_protocol`.
@@ -1039,6 +1039,71 @@ pub fn validate_sync_group_shape(version: i16, body: &Bytes, max_frame_size: usi
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`.
@@ -1128,7 +1193,7 @@ pub fn validate_sasl_authenticate_shape(version: i16, body: &Bytes) -> Result<()
#[cfg(test)]
mod tests {
use bytes::BytesMut;
- use kafka_protocol::messages::LeaveGroupRequest;
+ use kafka_protocol::messages::{DescribeGroupsRequest, LeaveGroupRequest, ListGroupsRequest};
use super::*;
use crate::protocol::handlers::decode_exhaustive;
@@ -1623,4 +1688,56 @@ mod tests {
assert!(validate_leave_group_shape(4, &body, 8 * 1024 * 1024).is_ok());
assert!(validate_leave_group_shape(4, &body, 1_024).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());
+ }
}
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..660b21b9a
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs
@@ -0,0 +1,142 @@
+// 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::{GroupDescription, MemberDescription};
+use crate::protocol::api::{
+ API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ERROR_GROUP_ID_NOT_FOUND, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_describe_groups_shape;
+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) {
+ return unsupported_version_response(API_KEY_DESCRIBE_GROUPS, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ 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 group_ids: Vec<StrBytes> = request
+ .groups
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let described = state.groups.describe_groups(&group_ids).await;
+ let groups = group_ids
+ .iter()
+ .zip(described)
+ .map(|(group_id, group)| {
+ group.map_or_else(|| missing_group(api_version, group_id), described_group)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, groups), "DescribeGroups")
+}
+
+fn described_group(group: GroupDescription) -> DescribedGroup {
+ 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())
+}
+
+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)
+}
+
+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)
+}
+
+/// # 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> {
+ // No top-level error field. Versions outside 0-6 fail in the encoder, which is what makes
+ // `unsupported_version_response` close the connection. A version this encoder accepts is
+ // not a version the firewall rejects.
+ encode_response(version, Vec::new())
+}
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..78f3250c7
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/list_groups.rs
@@ -0,0 +1,114 @@
+// 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,
+};
+
+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 groups = state
+ .groups
+ .list_groups()
+ .await
+ .into_iter()
+ .filter(|group| {
+ selected(&request.states_filter, group.state)
+ && selected(&request.types_filter, CLASSIC_GROUP_TYPE)
+ })
+ .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))
+}
+
+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)
+}
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 4ecfbe9fe..9852e4a4f 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -26,12 +26,14 @@
pub mod api_versions;
pub mod create_topics;
+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 produce;
@@ -44,11 +46,12 @@ use tokio::runtime::{Handle, RuntimeFlavor};
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, 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_PRODUCE, API_KEY_SYNC_GROUP, ConnectionState,
- ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
- is_supported_version, supported_max_version,
+ API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, 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_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.
@@ -91,6 +94,8 @@ pub async fn dispatch(
API_KEY_HEARTBEAT => heartbeat::handle(state, 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/tests/common/scope.rs b/gateways/kafka/tests/common/scope.rs
index 2926b0751..02feca860 100644
--- a/gateways/kafka/tests/common/scope.rs
+++ b/gateways/kafka/tests/common/scope.rs
@@ -35,6 +35,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),
(22, "InitProducerId", 0, 5),
diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs
index 89db5f219..6e4019f21 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -34,8 +34,6 @@ use super::codec::Encoder;
pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[
(8, "OffsetCommit"),
(9, "OffsetFetch"),
- (15, "DescribeGroups"),
- (16, "ListGroups"),
(20, "DeleteTopics"),
];
@@ -463,7 +461,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 cb10c6608..13e986b07 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,8 +48,9 @@ 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,
@@ -60,8 +61,9 @@ 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";
@@ -1416,6 +1418,349 @@ 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, "range");
+ assert_eq!(group.authorized_operations, Some(i32::MIN));
+ assert_eq!(group.members.len(), 2);
+ assert!(
+ group
+ .members
+ .iter()
+ .any(|member| member.metadata.as_ref() == b"leader-subscription")
+ );
+ assert!(
+ group
+ .members
+ .iter()
+ .any(|member| member.metadata.as_ref() == b"follower-subscription")
+ );
+ 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.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].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_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());
+}
+
// ── Over a real TCP listener ────────────────────────────────────────────────
/// The in-process tests drive one `GatewayState` directly. This one proves the connection loop
diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
index e99b77584..d6fc933ba 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=12 (compact array: N+1=13)
+ // error_code=0, api_count=14 (compact array: N+1=15)
// each entry followed by an empty tagged-fields byte; throttle_ms=0; top-level tagged fields
- let expected: [u8; 92] = [
+ let expected: [u8; 106] = [
0x00, 0x00, // error_code
- 0x0D, // compact array count (12+1)
+ 0x0F, // compact array count (14+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
@@ -55,6 +55,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, 0x16, 0x00, 0x00, 0x00, 0x05, 0x00, // key 22: InitProducerId 0-5
@@ -71,10 +73,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=12; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
- let expected: [u8; 82] = [
+ // error_code=0, api_count=14; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
+ let expected: [u8; 94] = [
0x00, 0x00, // error_code
- 0x00, 0x00, 0x00, 0x0C, // api count = 12
+ 0x00, 0x00, 0x00, 0x0E, // api count = 14
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
@@ -84,6 +86,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, 0x16, 0x00, 0x00, 0x00, 0x05, // key 22: InitProducerId 0-5
diff --git a/gateways/kafka/tests/server_e2e_tests.rs b/gateways/kafka/tests/server_e2e_tests.rs
index d12764040..4370e6612 100644
--- a/gateways/kafka/tests/server_e2e_tests.rs
+++ b/gateways/kafka/tests/server_e2e_tests.rs
@@ -422,7 +422,7 @@ async fn metadata_empty_body_e2e_closes_connection() {
async fn out_of_scope_api_keys_e2e_close() {
let (addr, _shutdown) = spawn_test_server().await;
- for &(api_key, name) in &OUT_OF_SCOPE_API_KEYS[..4] {
+ for &(api_key, name) in OUT_OF_SCOPE_API_KEYS {
let mut stream = TcpStream::connect(addr).await.expect("connect");
let frame = build_request_frame(api_key, 0, i32::from(api_key), Some("scope-test"), &[]);
stream.write_all(&frame).await.expect("write oos key");
diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs
index 63199e200..476bfa1c9 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -37,11 +37,12 @@ use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use iggy_gateway_kafka::protocol::api::{
- API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, 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_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_API_VERSIONS, API_KEY_CREATE_TOPICS, 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_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;
@@ -55,16 +56,17 @@ use tcp::{
};
use wire::{
JoinGroupParams, OUT_OF_SCOPE_API_KEYS, SyncGroupParams, build_api_versions_flexible_request,
- build_create_topics_empty_request, build_fetch_empty_topics_request,
- build_find_coordinator_request, build_heartbeat_request, build_init_producer_id_request,
- build_join_group_request, build_leave_group_request, build_list_offsets_request,
- build_metadata_all_topics_flexible, build_metadata_all_topics_legacy,
- build_metadata_flexible_request_v10, build_sync_group_request,
+ build_create_topics_empty_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_sync_group_request,
};
#[test]
-fn supported_ranges_table_has_twelve_entries() {
- assert_eq!(supported_api_ranges().len(), 12);
+fn supported_ranges_table_has_fourteen_entries() {
+ assert_eq!(supported_api_ranges().len(), 14);
}
#[test]
@@ -101,7 +103,8 @@ fn is_supported_version_matches_scope_table() {
///
/// Relies on `SUPPORTED_RANGES` (src) and `SCOPED_API_KEYS` (test) sharing declaration order
/// (Produce, Fetch, `ListOffsets`, Metadata, `ApiVersions`, `CreateTopics`, `FindCoordinator`,
-/// `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`) - `supported_ranges_table_has_twelve_entries` plus
+/// `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`, `DescribeGroups`, `ListGroups`) -
+/// `supported_ranges_table_has_fourteen_entries` plus
/// `is_supported_version_matches_scope_table` already pin that both tables cover the same keys.
#[tokio::test]
async fn apiversions_advertises_exact_supported_ranges_v1() {
@@ -518,6 +521,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, &[], &[]),
_ => Bytes::new(),
}
}
(iggy) 02/02: Intial version of Describe and List Groups
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch feat/gateways-3548-describe-list-groups
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 356839c0210753517e46c2dd2b370c364d3fd510
Author: ryerraguntla <ra...@gmail.com>
AuthorDate: Sun Oct 4 20:36:39 2026 -0400
Intial version of Describe and List Groups
---
gateways/kafka/docs/kafka_api_keys_reference.md | 15 +-
gateways/kafka/src/group/mod.rs | 40 +-
gateways/kafka/src/group/state.rs | 450 ++++++++++++++++++++-
.../kafka/src/protocol/handlers/describe_groups.rs | 40 +-
.../kafka/src/protocol/handlers/list_groups.rs | 78 +++-
gateways/kafka/tests/consumer_group_tests.rs | 164 +++++++-
6 files changed, 741 insertions(+), 46 deletions(-)
diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md
index 8c5582847..745e11295 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+)
@@ -269,7 +272,8 @@ Key new minimums:
| ---------- | :-----: | ------- |
| 🔴 Bridge (data path) | 7 | Produce, Fetch, Metadata, SaslHandshake, ApiVersions, SaslAuthenticate, ShareFetch |
| 🟠 Required Stub (client state machine) | 13 | ListOffsets, consumer group (8-14), CreateTopics, InitProducerId (22), ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) |
-| 🟡 Optional Stub (admin/observability) | 41 | Can return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` safely |
+| 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 |
@@ -294,7 +298,7 @@ Key new minimums:
| LeaveGroup | v0-v5 | v5 | current |
| SyncGroup | v0-v5 | v5 | current |
-### Missing from `SUPPORTED_RANGES` (76 of the 88 API keys in this document)
+### Missing from `SUPPORTED_RANGES` (74 of the 88 API keys in this document)
Every key not in `SUPPORTED_RANGES` closes the connection - the same policy applied to every
other unlisted key, not a special case for these. The gateway declines to define a response for a
@@ -308,7 +312,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).
Remaining scope (consumer groups, auth, admin/tuning) is tracked in `SCOPE.md`'s TODO section,
not duplicated here.
diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index b445d2120..16aefd5d4 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -27,7 +27,7 @@
mod state;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::time::Duration;
use bytes::Bytes;
@@ -39,6 +39,7 @@ use tokio_util::sync::CancellationToken;
use crate::group::state::{GroupState, Step};
use crate::protocol::api::{ERROR_NONE, ERROR_NOT_COORDINATOR, ERROR_UNKNOWN_MEMBER_ID};
+pub use state::DescribeGroupsView;
/// `DescribeGroups` / `ListGroups` state string for a group preparing a rebalance.
pub const GROUP_STATE_PREPARING_REBALANCE: &str = "PreparingRebalance";
@@ -488,13 +489,27 @@ impl GroupCoordinator {
state::leave_step(&mut groups, request, Instant::now())
}
- /// Snapshot of each named group, in request order. `None` means the group is not here.
+ /// Snapshot of each distinct group, in first-seen order. `None` means the group is not here.
///
- /// Ticks each named group first, the same way every other request for that group does, so the
- /// snapshot is the membership a heartbeat arriving now would see.
- pub async fn describe_groups(&self, group_ids: &[StrBytes]) -> Vec<Option<GroupDescription>> {
+ /// Ids are deduped before the lock is taken. Under the lock each remaining group is ticked
+ /// once and, when the encoded response fits `max_frame_size`, snapshotted once. Encoding
+ /// happens after this returns. `ExceedsFrame` means the lengths already stored would not
+ /// fit, and no member snapshot was built.
+ pub async fn describe_groups(
+ &self,
+ group_ids: &[StrBytes],
+ version: i16,
+ max_frame_size: usize,
+ ) -> DescribeGroupsView {
+ let group_ids = distinct_group_ids(group_ids);
let mut groups = self.groups.lock().await;
- state::describe_groups(&mut groups, group_ids, Instant::now())
+ state::describe_groups(
+ &mut groups,
+ &group_ids,
+ version,
+ max_frame_size,
+ Instant::now(),
+ )
}
/// Every group currently in the map, ordered by group id.
@@ -522,6 +537,19 @@ enum Parked<T> {
Wait(StrBytes, Instant, watch::Receiver<u64>),
}
+/// First-seen order, one entry per id. `DescribeGroups` answers a repeated id once: a second
+/// snapshot of the same members would multiply the encoded response by the repeat count.
+pub(crate) fn distinct_group_ids(group_ids: &[StrBytes]) -> Vec<StrBytes> {
+ let mut seen = HashSet::with_capacity(group_ids.len());
+ let mut distinct = Vec::with_capacity(group_ids.len());
+ for group_id in group_ids {
+ if seen.insert(group_id.clone()) {
+ distinct.push(group_id.clone());
+ }
+ }
+ distinct
+}
+
fn park_outcome<T>(
groups: &HashMap<StrBytes, GroupState>,
group_id: &StrBytes,
diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs
index 8bfcb9b89..16cf07688 100644
--- a/gateways/kafka/src/group/state.rs
+++ b/gateways/kafka/src/group/state.rs
@@ -41,11 +41,12 @@ use crate::group::{
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,
};
+use crate::protocol::header::response_header_version;
/// 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`).
@@ -55,6 +56,15 @@ pub const MAX_GROUP_ID_BYTES: usize = 246;
/// 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
+/// `protocol_type`, `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;
+
+/// `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";
@@ -184,6 +194,40 @@ 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.
+ fn described_len(
+ &self,
+ version: i16,
+ member_id: &StrBytes,
+ protocol_name: Option<&StrBytes>,
+ ) -> 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 = protocol_name.map_or(0, |name| self.metadata_len(name));
+ total += kafka_bytes_len(flexible, metadata_len);
+ total += kafka_bytes_len(flexible, self.assignment.len());
+ if flexible {
+ total += 1;
+ }
+ total
+ }
}
pub struct GroupState {
@@ -654,6 +698,38 @@ impl GroupState {
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) -> usize {
+ 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.phase.as_kafka_state().len());
+ total += kafka_string_len(flexible, self.protocol_type.len());
+ total += kafka_string_len(
+ flexible,
+ self.protocol_name.as_ref().map_or(0, |name| name.len()),
+ );
+ total += kafka_array_prefix(flexible, self.members.len());
+ for (member_id, member) in &self.members {
+ total = total.saturating_add(member.described_len(
+ version,
+ member_id,
+ self.protocol_name.as_ref(),
+ ));
+ }
+ if version >= 3 {
+ total += 4;
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+ }
}
/// Tick every group and drop the ones that emptied, returning how many were reclaimed.
@@ -723,7 +799,7 @@ 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
@@ -743,6 +819,26 @@ fn join_request_error(config: &GroupCoordinatorConfig, request: &JoinRequest) ->
None
}
+/// `protocol_type`, a client-supplied `member_id`, and `group_instance_id` are echoed as legacy
+/// strings. When the member id is generated, the instance id is the prefix and the `-{uuid}`
+/// suffix has to fit in the same cap.
+fn legacy_strings_too_long(request: &JoinRequest) -> bool {
+ if request.protocol_type.len() > MAX_LEGACY_STRING_BYTES
+ || request.member_id.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
@@ -1147,25 +1243,44 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
-/// One entry per `group_ids` element, after that group's usual expiry tick.
+/// What [`describe_groups`] decided under the coordinator lock.
+///
+/// `ExceedsFrame` means the encoded response, header included, would pass `max_frame_size`. No
+/// member snapshot was built: the lengths already stored on the group are enough to price it.
+pub enum DescribeGroupsView {
+ Groups(Vec<Option<GroupDescription>>),
+ ExceedsFrame { encoded_bytes: usize },
+}
+
+/// 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. The frame check runs on those lengths before [`GroupState::description`], which is
+/// what copies member metadata and assignments.
pub fn describe_groups(
groups: &mut Groups,
group_ids: &[StrBytes],
+ version: i16,
+ max_frame_size: usize,
now: Instant,
-) -> Vec<Option<GroupDescription>> {
- let mut described = Vec::with_capacity(group_ids.len());
- for group_id in group_ids {
- if !tick_group(groups, group_id, now) {
- described.push(None);
- continue;
- }
- let Some(group) = groups.get(group_id) else {
- described.push(None);
- continue;
- };
- described.push(Some(group.description(group_id)));
+) -> DescribeGroupsView {
+ let group_ids = super::distinct_group_ids(group_ids);
+ for group_id in &group_ids {
+ tick_group(groups, group_id, now);
+ }
+ let encoded_bytes = describe_frame_len(version, groups, &group_ids);
+ if encoded_bytes > max_frame_size {
+ return DescribeGroupsView::ExceedsFrame { encoded_bytes };
}
- described
+ let described = group_ids
+ .iter()
+ .map(|group_id| {
+ groups
+ .get(group_id)
+ .map(|group| group.description(group_id))
+ })
+ .collect();
+ DescribeGroupsView::Groups(described)
}
/// Groups currently in the map, ordered by group id. Does not tick: see
@@ -1183,6 +1298,107 @@ pub fn list_groups(groups: &Groups) -> Vec<GroupListing> {
listed
}
+/// Header plus body, the length `send_response` writes and `read_frame` compares to
+/// `max_frame_size`. The 4-byte length prefix is not part of that value.
+fn describe_frame_len(version: i16, groups: &Groups, group_ids: &[StrBytes]) -> 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_ids.len());
+ for group_id in group_ids {
+ let group_bytes = groups.get(group_id).map_or_else(
+ || missing_described_len(version, group_id),
+ |group| group.described_len(version, group_id),
+ );
+ total = total.saturating_add(group_bytes);
+ }
+ if version >= 5 {
+ 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
+ }
+}
+
+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`.
+fn kafka_compact_prefix(len: usize) -> usize {
+ let wire = u32::try_from(len.saturating_add(1)).unwrap_or(u32::MAX);
+ unsigned_varint_len(wire)
+}
+
+fn kafka_string_len(flexible: bool, bytes: usize) -> usize {
+ if flexible {
+ kafka_compact_prefix(bytes) + bytes
+ } else {
+ 2 + bytes
+ }
+}
+
+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
+ }
+}
+
fn admit(
group: &mut GroupState,
config: &GroupCoordinatorConfig,
@@ -1252,7 +1468,15 @@ fn generate_member_id(group_instance_id: Option<&StrBytes>) -> StrBytes {
#[cfg(test)]
mod tests {
+ use bytes::BytesMut;
+ use kafka_protocol::messages::describe_groups_response::{
+ DescribedGroup, DescribedGroupMember,
+ };
+ use kafka_protocol::messages::{DescribeGroupsResponse, GroupId};
+ use kafka_protocol::protocol::Encodable;
+
use super::*;
+ use crate::protocol::api::ERROR_GROUP_ID_NOT_FOUND;
const GROUP: &str = "g";
@@ -1631,6 +1855,194 @@ mod tests {
);
}
+ /// `DescribeGroups` v0-v4 and `ListGroups` v0-v2 encode `protocol_type` as a legacy string.
+ /// Join v6 can carry one longer than `i16::MAX`, and storing it would make those responses
+ /// unencodable.
+ #[test]
+ fn given_a_protocol_type_past_the_legacy_string_cap_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_LEGACY_STRING_BYTES + 1));
+
+ let step = join_step(&mut groups, &config, &request, Instant::now());
+
+ assert_eq!(error_of(&step), ERROR_INVALID_REQUEST);
+ assert!(
+ groups.is_empty(),
+ "a protocol_type that cannot be echoed must not be stored"
+ );
+ }
+
+ /// 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, and a frame of 0 bytes must refuse
+ /// before that snapshot exists. The group stays: refusing the response is not a deletion.
+ #[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);
+ let DescribeGroupsView::Groups(described) = described else {
+ panic!("a single small group fits any real frame");
+ };
+ assert_eq!(described.len(), 1);
+ assert!(described[0].is_some());
+
+ let refused = describe_groups(&mut groups, std::slice::from_ref(&id), 0, 0, now);
+ assert!(matches!(refused, DescribeGroupsView::ExceedsFrame { .. }));
+ assert!(
+ groups.contains_key(&id),
+ "closing the describe response must not drop the group"
+ );
+ }
+
+ /// The frame check has to match what `DescribeGroups` actually encodes, header included.
+ #[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 _ = join_step(&mut groups, &config, &request("", &["range"]), now);
+ let id = group_id();
+ let missing = StrBytes::from_static_str("missing");
+
+ for version in 0..=6 {
+ for ids in [
+ vec![id.clone()],
+ vec![missing.clone()],
+ vec![id.clone(), missing.clone()],
+ ] {
+ let priced = describe_frame_len(version, &groups, &ids);
+ let DescribeGroupsView::Groups(described) =
+ describe_groups(&mut groups, &ids, version, usize::MAX, now)
+ else {
+ panic!("version {version} must fit");
+ };
+ let encoded = encode_described(version, &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 encoded_described_group(group: &GroupDescription) -> DescribedGroup {
+ DescribedGroup::default()
+ .with_group_id(GroupId(group.group_id.clone()))
+ .with_group_state(StrBytes::from_static_str(group.state))
+ .with_protocol_type(group.protocol_type.clone())
+ .with_protocol_data(group.protocol_name.clone().unwrap_or_default())
+ .with_members(
+ group
+ .members
+ .iter()
+ .map(|member| {
+ DescribedGroupMember::default()
+ .with_member_id(member.member_id.clone())
+ .with_group_instance_id(member.group_instance_id.clone())
+ .with_member_metadata(member.metadata.clone())
+ .with_member_assignment(member.assignment.clone())
+ })
+ .collect(),
+ )
+ }
+
+ fn missing_described(version: i16, group_id: &StrBytes) -> DescribedGroup {
+ let mut missing = DescribedGroup::default()
+ .with_group_id(GroupId(group_id.clone()))
+ .with_group_state(StrBytes::from_static_str("Dead"));
+ if version >= 6 {
+ missing = missing
+ .with_error_code(ERROR_GROUP_ID_NOT_FOUND)
+ .with_error_message(Some(StrBytes::from_string(format!(
+ "Group {} not found.",
+ group_id.as_str()
+ ))));
+ }
+ missing
+ }
+
+ fn encode_described(
+ version: i16,
+ ids: &[StrBytes],
+ described: &[Option<GroupDescription>],
+ ) -> Bytes {
+ let groups = ids
+ .iter()
+ .zip(described)
+ .map(|(group_id, group)| {
+ group.as_ref().map_or_else(
+ || missing_described(version, group_id),
+ encoded_described_group,
+ )
+ })
+ .collect();
+ let response = DescribeGroupsResponse::default().with_groups(groups);
+ let mut buf = BytesMut::new();
+ response
+ .encode(&mut buf, version)
+ .expect("described group encodes");
+ buf.freeze()
+ }
+
/// 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/handlers/describe_groups.rs b/gateways/kafka/src/protocol/handlers/describe_groups.rs
index 660b21b9a..0dbf84ab3 100644
--- a/gateways/kafka/src/protocol/handlers/describe_groups.rs
+++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs
@@ -31,7 +31,7 @@ use kafka_protocol::messages::{DescribeGroupsRequest, DescribeGroupsResponse, Gr
use kafka_protocol::protocol::StrBytes;
use crate::error::Result;
-use crate::group::{GroupDescription, MemberDescription};
+use crate::group::{DescribeGroupsView, GroupDescription, MemberDescription, distinct_group_ids};
use crate::protocol::api::{
API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ERROR_GROUP_ID_NOT_FOUND, ERROR_UNSUPPORTED_VERSION,
GatewayState, HandleOutcome, is_supported_version,
@@ -73,20 +73,38 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
}
};
- let group_ids: Vec<StrBytes> = request
+ let requested: Vec<StrBytes> = request
.groups
.iter()
.map(|group_id| group_id.0.clone())
.collect();
- let described = state.groups.describe_groups(&group_ids).await;
- let groups = group_ids
- .iter()
- .zip(described)
- .map(|(group_id, group)| {
- group.map_or_else(|| missing_group(api_version, group_id), described_group)
- })
- .collect();
- respond_or_close(encode_response(api_version, groups), "DescribeGroups")
+ // 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 = distinct_group_ids(&requested);
+ match state
+ .groups
+ .describe_groups(&group_ids, api_version, state.max_frame_size)
+ .await
+ {
+ DescribeGroupsView::ExceedsFrame { encoded_bytes } => {
+ tracing::warn!(
+ encoded_bytes,
+ max_frame_size = state.max_frame_size,
+ "DescribeGroups response would exceed max_frame_size; closing connection"
+ );
+ HandleOutcome::Close
+ }
+ DescribeGroupsView::Groups(described) => {
+ let groups = group_ids
+ .iter()
+ .zip(described)
+ .map(|(group_id, group)| {
+ group.map_or_else(|| missing_group(api_version, group_id), described_group)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, groups), "DescribeGroups")
+ }
+ }
}
fn described_group(group: GroupDescription) -> DescribedGroup {
diff --git a/gateways/kafka/src/protocol/handlers/list_groups.rs b/gateways/kafka/src/protocol/handlers/list_groups.rs
index 78f3250c7..5a78c045f 100644
--- a/gateways/kafka/src/protocol/handlers/list_groups.rs
+++ b/gateways/kafka/src/protocol/handlers/list_groups.rs
@@ -38,6 +38,7 @@ 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;
pub const RANGE: ApiVersionRange = ApiVersionRange {
api_key: API_KEY_LIST_GROUPS,
@@ -68,7 +69,7 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
}
};
- let groups = state
+ let listings: Vec<GroupListing> = state
.groups
.list_groups()
.await
@@ -77,8 +78,19 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
selected(&request.states_filter, group.state)
&& selected(&request.types_filter, CLASSIC_GROUP_TYPE)
})
- .map(listed_group)
.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. Price it before encode.
+ let encoded_bytes = list_response_frame_len(api_version, &listings);
+ if encoded_bytes > state.max_frame_size {
+ tracing::warn!(
+ encoded_bytes,
+ max_frame_size = state.max_frame_size,
+ "ListGroups response would exceed max_frame_size; closing connection"
+ );
+ return HandleOutcome::Close;
+ }
+ let groups = listings.into_iter().map(listed_group).collect();
respond_or_close(encode_response(api_version, groups), "ListGroups")
}
@@ -89,6 +101,68 @@ fn selected(filter: &[StrBytes], value: &str) -> bool {
.any(|wanted| wanted.as_str().trim().eq_ignore_ascii_case(value))
}
+/// 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 kafka_compact_prefix(len: usize) -> usize {
+ let wire = u32::try_from(len.saturating_add(1)).unwrap_or(u32::MAX);
+ match wire {
+ 0x00..=0x7f => 1,
+ 0x80..=0x3fff => 2,
+ 0x4000..=0x001f_ffff => 3,
+ 0x0020_0000..=0x0fff_ffff => 4,
+ _ => 5,
+ }
+}
+
+fn kafka_string_len(flexible: bool, bytes: usize) -> usize {
+ if flexible {
+ kafka_compact_prefix(bytes) + bytes
+ } else {
+ 2 + bytes
+ }
+}
+
fn listed_group(group: GroupListing) -> ListedGroup {
ListedGroup::default()
.with_group_id(GroupId(group.group_id))
diff --git a/gateways/kafka/tests/consumer_group_tests.rs b/gateways/kafka/tests/consumer_group_tests.rs
index 13e986b07..e511e6f43 100644
--- a/gateways/kafka/tests/consumer_group_tests.rs
+++ b/gateways/kafka/tests/consumer_group_tests.rs
@@ -56,6 +56,7 @@ use iggy_gateway_kafka::protocol::api::{
ERROR_REBALANCE_IN_PROGRESS, ERROR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED,
ERROR_UNKNOWN_MEMBER_ID, GatewayState, handle_request_bounded,
};
+use iggy_gateway_kafka::protocol::header::response_header_version;
use codec::Decoder;
use server::spawn_test_server_with_config;
@@ -77,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
@@ -1761,6 +1775,150 @@ async fn given_an_unknown_group_when_describing_should_report_dead_without_creat
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_at_the_frame_cap_when_one_byte_over_should_close() {
+ let mut state = owned_state(immediate_config(), 8 * 1024 * 1024);
+ // Bigger than the shape guard's per-id charge, so a repeated id is not rejected as a
+ // malformed request before the response-size check can see that it is one group.
+ let blob = vec![b'x'; 400];
+ let protocols: &[(&str, &[u8])] = &[("range", blob.as_slice())];
+ join(&state, 3, &join_params("", protocols)).await;
+
+ 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"
+ );
+
+ state.max_frame_size = frame - 1;
+ assert!(
+ handle_request_bounded(
+ &state,
+ API_KEY_DESCRIBE_GROUPS,
+ version,
+ build_describe_groups_request(version, &[GROUP], false),
+ )
+ .await
+ .is_close(),
+ "DescribeGroups closes when the encoded frame would pass max_frame_size"
+ );
+
+ 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_ne!(again.groups[0].state, "Dead");
+}
+
+#[tokio::test(start_paused = true)]
+async fn given_a_list_response_one_byte_past_the_frame_cap_when_listing_should_close() {
+ 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);
+
+ state.max_frame_size = frame - 1;
+ assert!(
+ handle_request_bounded(
+ &state,
+ API_KEY_LIST_GROUPS,
+ version,
+ build_list_groups_request(version, &[], &[]),
+ )
+ .await
+ .is_close(),
+ "ListGroups closes 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