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 <[email protected]> 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(), } }
