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 <[email protected]> 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
