ryerraguntla commented on code in PR #4263: URL: https://github.com/apache/iggy/pull/4263#discussion_r4100413942
########## gateways/kafka/src/protocol/handlers/sync_group.rs: ########## @@ -0,0 +1,86 @@ +// 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. + +//! `SyncGroup` (API key 14). +//! +//! The leader ships one opaque blob per member and every member is handed back exactly the bytes +//! the leader filed under its own id, in the same generation. That relay is the whole contract: +//! partitions are disjoint because the leader's assignor made them so, not because this gateway +//! looked inside. + +use bytes::Bytes; +use kafka_protocol::messages::{SyncGroupRequest, SyncGroupResponse}; + +use crate::error::Result; +use crate::group::{SyncRequest, SyncResult}; +use crate::protocol::api::{ + API_KEY_SYNC_GROUP, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, + GatewayState, HandleOutcome, is_supported_version, +}; +use crate::protocol::bounds_guard::validate_sync_group_shape; +use crate::protocol::handlers::{ + decode_guarded, encode_message, respond_or_close, unsupported_version_response, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_SYNC_GROUP, + min_version: 0, + max_version: 5, +}; + +pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { + if !is_supported_version(API_KEY_SYNC_GROUP, api_version) { + return unsupported_version_response(API_KEY_SYNC_GROUP, api_version, |version| { + encode_error_response(version, ERROR_UNSUPPORTED_VERSION) + }); + } + let request = match decode_guarded::<SyncGroupRequest>(api_version, body, |version, body| { + validate_sync_group_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 SyncGroup request"); + return respond_or_close( + encode_error_response(api_version, ERROR_INVALID_REQUEST), + "SyncGroup", + ); + } + }; + + let result = state.groups.sync(&SyncRequest::from(&request)).await; Review Comment: decoded SyncGroupRequest lives across sync().await. SyncRequest::from clones StrBytes/Bytes (group/mod.rs:172) = inbound-frame refcount. Follower CompletingRebalance → Wait (state.rs:925). Leader silent. Park ≤30 min. Caps miss parked frame. Unauth: 999 waiters × 8 MiB ≈ 8 GiB. Join path already copies + drops. **Fix**: same owned_str + copy_from_slice; drop(request) before .await; drop follower assignments on Wait. ########## gateways/kafka/src/group/state.rs: ########## @@ -0,0 +1,2059 @@ +// 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. + +//! Synchronous state machine behind [`crate::group::GroupCoordinator`]. +//! +//! Every entry point takes the whole group map, a config, and `now`, and returns either a +//! response or a deadline to park until. Nothing here allocates a future or touches a clock, so +//! the protocol rules are testable without a runtime and the coordinator's lock is never held +//! across an `.await`. +//! +//! Kafka's `Empty` group state is "absent from the map": offsets live in Iggy, so an empty group +//! holds nothing worth keeping and retaining it would be an unbounded-memory vector. + +use std::collections::{BTreeMap, HashMap}; +use std::time::Duration; + +use bytes::Bytes; +use kafka_protocol::protocol::StrBytes; +use tokio::sync::watch; +use tokio::time::Instant; +use uuid::Uuid; + +use crate::group::{ + GroupCoordinatorConfig, JoinRequest, JoinResult, JoinedMember, SyncRequest, SyncResult, + owned_str, +}; +use crate::protocol::api::{ + ERROR_COORDINATOR_NOT_AVAILABLE, 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, +}; + +/// 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`). +pub const MAX_GROUP_ID_BYTES: usize = 246; + +/// 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"; + +/// Floor on how long a parked waiter sleeps. Every deadline a tick leaves behind is strictly in +/// the future, so this only guards against a future rule that forgets to maintain that and turns +/// a park into a spin. +const MIN_PARK: Duration = Duration::from_millis(1); + +pub type Groups = HashMap<StrBytes, GroupState>; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Phase { + PreparingRebalance, + CompletingRebalance, + Stable, +} + +/// Outcome of one state-machine step: answer now, or park until `wake_at`. +pub enum Step<T> { + Respond(T), + Wait { + member_id: StrBytes, + wake_at: Instant, + }, +} + +pub struct Member { + group_instance_id: Option<StrBytes>, + session_timeout: Duration, + rebalance_timeout: Duration, + protocols: Vec<(StrBytes, Bytes)>, + assignment: Bytes, + session_deadline: Instant, + /// Has this member rejoined in the rebalance currently being prepared? + rejoined: bool, + /// Has this member sent `SyncGroup` in the current generation? + synced: bool, + /// Taken by the member's own parked `JoinGroup` handler. Snapshotting the answer at join + /// completion means a rebalance that starts before the waiter wakes cannot change what the + /// member is told about the generation it just completed. + join_response: Option<JoinResult>, +} + +/// Copy what a member keeps out of the request frame. +/// +/// `StrBytes` and `Bytes` are refcounted views, so retaining them verbatim keeps the whole frame +/// alive: a member whose counted metadata is a few hundred bytes can pin megabytes. Copying costs +/// one allocation per protocol on a control-plane path, and bounds what a member actually retains +/// to what the caps actually measure. +fn retained_protocols(protocols: &[(StrBytes, Bytes)]) -> Vec<(StrBytes, Bytes)> { + protocols + .iter() + .map(|(name, metadata)| (owned_str(name), Bytes::copy_from_slice(metadata))) + .collect() +} + +impl Member { + fn new(request: &JoinRequest, now: Instant, max_rebalance_timeout: Duration) -> Self { + Self { + group_instance_id: request.group_instance_id.as_ref().map(owned_str), + session_timeout: request.session_timeout, + rebalance_timeout: request.rebalance_timeout.min(max_rebalance_timeout), + protocols: retained_protocols(&request.protocols), + assignment: Bytes::new(), + session_deadline: now + request.session_timeout, + rejoined: true, + synced: false, + join_response: None, + } + } + + fn rejoin(&mut self, request: &JoinRequest, now: Instant, max_rebalance_timeout: Duration) { + self.group_instance_id = request.group_instance_id.as_ref().map(owned_str); + self.session_timeout = request.session_timeout; + self.rebalance_timeout = request.rebalance_timeout.min(max_rebalance_timeout); + self.protocols = retained_protocols(&request.protocols); + self.session_deadline = now + request.session_timeout; + self.rejoined = true; + // Cleared here rather than when a rebalance opens: a member that has rejoined is asking + // for the next generation's answer, while one still parked on the previous generation + // must keep the snapshot it is waiting to collect. + self.join_response = None; + } + + fn roster_bytes(&self, member_id: &StrBytes) -> usize { + roster_entry_bytes(member_id, self.group_instance_id.as_ref(), &self.protocols) + } + + fn supports(&self, name: &StrBytes) -> bool { + self.protocols + .iter() + .any(|(candidate, _)| candidate == name) + } + + /// This member's vote: the first protocol it listed that the whole group can speak. + fn vote<'a>(&self, candidates: &'a [StrBytes]) -> Option<&'a StrBytes> { + self.protocols + .iter() + .find_map(|(name, _)| candidates.iter().find(|candidate| *candidate == name)) + } + + fn metadata_for(&self, name: &StrBytes) -> Bytes { + self.protocols + .iter() + .find(|(candidate, _)| candidate == name) + .map_or_else(Bytes::new, |(_, metadata)| metadata.clone()) + } +} + +pub struct GroupState { + phase: Phase, + /// A group that has completed one rebalance is at generation 1. + generation_id: i32, + protocol_type: StrBytes, + protocol_name: Option<StrBytes>, + leader: Option<StrBytes>, + /// Ordered, not hashed: iteration decides leader fallback and protocol tie-breaks. + members: BTreeMap<StrBytes, Member>, + /// Ids handed out with `MEMBER_ID_REQUIRED` that have not rejoined yet, and their expiry. + pending: BTreeMap<StrBytes, Instant>, + join_deadline: Option<Instant>, + sync_deadline: Option<Instant>, + rebalance_started: Instant, + /// First join of a new group: the barrier waits out the full delay even once every known + /// member has joined, so a second consumer starting a moment later lands in generation 1. + initial: bool, + changed: watch::Sender<u64>, +} + +impl GroupState { + /// A new group's join window does not open here: it opens when the first member is admitted + /// (`admit`). A `MEMBER_ID_REQUIRED` reply creates the group but adds no member, and a window + /// opened at that moment would expire while the client was still on its way back with the id + /// it was just given. + fn new(protocol_type: StrBytes, now: Instant) -> Self { + Self { + phase: Phase::PreparingRebalance, + generation_id: 0, + protocol_type, + protocol_name: None, + leader: None, + members: BTreeMap::new(), + pending: BTreeMap::new(), + join_deadline: None, + sync_deadline: None, + rebalance_started: now, + initial: true, + changed: watch::channel(0).0, + } + } + + pub fn subscribe(&self) -> watch::Receiver<u64> { + self.changed.subscribe() + } + + /// `send_modify`, never `send`: the latter errors once the last receiver is gone, which is + /// the normal state of a group whose members are all between requests. + fn bump(&self) { + self.changed + .send_modify(|version| *version = version.wrapping_add(1)); + } + + fn is_empty(&self) -> bool { + self.members.is_empty() && self.pending.is_empty() + } + + /// Would admitting `request` as `member_id` grow the leader's roster past its cap? The + /// member's own current entry is replaced, not added to. + fn roster_overflows( + &self, + config: &GroupCoordinatorConfig, + member_id: &StrBytes, + request: &JoinRequest, + ) -> bool { + let others: usize = self + .members + .iter() + .filter(|(id, _)| *id != member_id) + .map(|(id, member)| member.roster_bytes(id)) + .sum(); + let joining = roster_entry_bytes( + member_id, + request.group_instance_id.as_ref(), + &request.protocols, + ); + others + joining > config.max_group_roster_bytes + } + + fn max_rebalance_timeout(&self) -> Duration { + self.members + .values() + .map(|member| member.rebalance_timeout) + .max() + .unwrap_or(Duration::ZERO) + } + + /// Expire whatever is overdue and complete whichever phase that unblocks. + /// + /// Expiry is judged against the recorded deadline, not against when this happens to run, so + /// a request arriving after its own member's deadline finds that member already gone. That + /// is stricter than a broker, whose timer thread may not have fired yet, and it is what + /// makes eviction deterministic here. + fn tick(&mut self, now: Instant) { + let mut changed = false; + + let expired_pending: Vec<StrBytes> = self + .pending + .iter() + .filter(|(_, deadline)| **deadline <= now) + .map(|(id, _)| id.clone()) + .collect(); + for id in &expired_pending { + self.pending.remove(id); + } + changed |= !expired_pending.is_empty(); + + let expired: Vec<StrBytes> = self + .members + .iter() + .filter(|(_, member)| member.session_deadline <= now) + .map(|(id, _)| id.clone()) + .collect(); + for id in &expired { + self.remove_member(id); + } + if !expired.is_empty() { + changed = true; + if !self.members.is_empty() && self.phase != Phase::PreparingRebalance { + self.prepare_rebalance(now, None); + } + } + + if self.phase == Phase::PreparingRebalance { + self.maybe_complete_join(now); + } + + if self.phase == Phase::CompletingRebalance + && self.sync_deadline.is_some_and(|deadline| deadline <= now) + { + let unsynced: Vec<StrBytes> = self + .members + .iter() + .filter(|(_, member)| !member.synced) + .map(|(id, _)| id.clone()) + .collect(); + for id in &unsynced { + self.remove_member(id); + } + self.sync_deadline = None; + if !self.members.is_empty() { + self.prepare_rebalance(now, None); + } + changed = true; + } + + if changed { + self.bump(); + } + } + + fn remove_member(&mut self, member_id: &StrBytes) { + self.members.remove(member_id); + if self.leader.as_ref() == Some(member_id) { + self.leader = self.members.keys().next().cloned(); + } + } + + /// The earliest moment any rule in this group could fire. + fn next_deadline(&self) -> Option<Instant> { + let phase_deadline = match self.phase { + Phase::PreparingRebalance => self.join_deadline, + Phase::CompletingRebalance => self.sync_deadline, + Phase::Stable => None, + }; + phase_deadline + .into_iter() + .chain(self.members.values().map(|member| member.session_deadline)) + .chain(self.pending.values().copied()) + .min() + } + + fn wake_at(&self, now: Instant) -> Instant { + let floor = now + MIN_PARK; + self.next_deadline().unwrap_or(floor).max(floor) + } + + fn prepare_rebalance(&mut self, now: Instant, trigger: Option<&StrBytes>) { + self.phase = Phase::PreparingRebalance; + self.initial = false; + self.rebalance_started = now; + self.sync_deadline = None; + self.join_deadline = Some(now + self.max_rebalance_timeout()); + for (member_id, member) in &mut self.members { + member.rejoined = trigger == Some(member_id); + } + self.bump(); + } + + fn maybe_complete_join(&mut self, now: Instant) { + let all_joined = !self.initial + && self.pending.is_empty() + && !self.members.is_empty() + && self.members.values().all(|member| member.rejoined); + if all_joined || self.join_deadline.is_some_and(|deadline| now >= deadline) { + self.complete_join(now); + } + } + + fn complete_join(&mut self, now: Instant) { + self.members.retain(|_, member| member.rejoined); + self.pending.clear(); + self.join_deadline = None; + self.initial = false; + if self.members.is_empty() { + self.leader = None; + self.bump(); + return; + } + + let leader = match self.leader.clone() { + Some(leader) if self.members.contains_key(&leader) => leader, + _ => self.members.keys().next().cloned().unwrap_or_default(), + }; + let protocol = self.select_protocol(&leader); + self.leader = Some(leader.clone()); + self.protocol_name.clone_from(&protocol); + self.generation_id = self.generation_id.wrapping_add(1); + self.phase = Phase::CompletingRebalance; + self.sync_deadline = Some(now + self.max_rebalance_timeout()); + + let selected = protocol.unwrap_or_default(); + let roster: Vec<JoinedMember> = self + .members + .iter() + .map(|(member_id, member)| JoinedMember { + member_id: member_id.clone(), + group_instance_id: member.group_instance_id.clone(), + metadata: member.metadata_for(&selected), + }) + .collect(); + + let generation_id = self.generation_id; + let protocol_type = self.protocol_type.clone(); + for (member_id, member) in &mut self.members { + member.session_deadline = now + member.session_timeout; + member.synced = false; + member.rejoined = false; + member.assignment = Bytes::new(); + member.join_response = Some(JoinResult { + error: ERROR_NONE, + generation_id, + protocol_type: Some(protocol_type.clone()), + protocol_name: Some(selected.clone()), + leader: leader.clone(), + member_id: member_id.clone(), + members: if *member_id == leader { + roster.clone() + } else { + Vec::new() + }, + }); + } + self.bump(); + } + + /// The protocol every member speaks, most first-preference votes winning. + /// + /// Candidates are walked in the leader's own listed order, which breaks a tie deterministically + /// where Kafka breaks it by set iteration order. + fn select_protocol(&self, leader_id: &StrBytes) -> Option<StrBytes> { + let leader = self.members.get(leader_id)?; + let candidates: Vec<StrBytes> = leader + .protocols + .iter() + .map(|(name, _)| name.clone()) + .filter(|name| self.members.values().all(|member| member.supports(name))) + .collect(); + + let mut best: Option<(StrBytes, usize)> = None; + for name in &candidates { + let votes = self + .members + .values() + .filter(|member| member.vote(&candidates) == Some(name)) + .count(); + if best.as_ref().is_none_or(|(_, most)| votes > *most) { + best = Some((name.clone(), votes)); + } + } + best.map(|(name, _)| name) + } + + /// Can `protocols` still leave one name every other member speaks? + fn protocols_compatible(&self, member_id: &StrBytes, protocols: &[(StrBytes, Bytes)]) -> bool { + protocols.iter().any(|(name, _)| { + self.members + .iter() + .all(|(id, member)| id == member_id || member.supports(name)) + }) + } + + fn current_generation_result(&self, member_id: &StrBytes) -> JoinResult { + // Only the leader runs an assignor, so only the leader is given the roster. Handing it an + // empty one would have it assign nothing to everybody and leave the group consuming no + // partitions, silently. + let members = if self.leader.as_deref() == Some(member_id.as_ref()) { + self.roster() + } else { + Vec::new() + }; + JoinResult { + error: ERROR_NONE, + generation_id: self.generation_id, + protocol_type: Some(self.protocol_type.clone()), + protocol_name: self.protocol_name.clone(), + leader: self.leader.clone().unwrap_or_default(), + member_id: member_id.clone(), + members, + } + } + + /// The member list an assignor needs, in the group's selected protocol. + fn roster(&self) -> Vec<JoinedMember> { + let selected = self.protocol_name.clone().unwrap_or_default(); + self.members + .iter() + .map(|(member_id, member)| JoinedMember { + member_id: member_id.clone(), + group_instance_id: member.group_instance_id.clone(), + metadata: member.metadata_for(&selected), + }) + .collect() + } + + fn sync_result(&self, member_id: &StrBytes) -> SyncResult { + SyncResult { + error: ERROR_NONE, + protocol_type: Some(self.protocol_type.clone()), + protocol_name: self.protocol_name.clone(), + assignment: self + .members + .get(member_id) + .map_or_else(Bytes::new, |member| member.assignment.clone()), + } + } + + /// Fan the leader's blobs out to every member. A member the leader left out gets empty bytes, + /// which is what a broker stores for it too. + fn apply_assignments(&mut self, assignments: &[(StrBytes, Bytes)], now: Instant) { + for (member_id, member) in &mut self.members { + member.assignment = assignments + .iter() + .find(|(target, _)| target == member_id) + .map_or_else(Bytes::new, |(_, blob)| Bytes::copy_from_slice(blob)); + member.session_deadline = now + member.session_timeout; + } + if let Some(leader) = self.leader.clone() + && let Some(member) = self.members.get_mut(&leader) + { + member.synced = true; + } + self.phase = Phase::Stable; + self.sync_deadline = None; + self.bump(); + } +} + +/// Tick every group and drop the ones that emptied, returning how many were reclaimed. +/// +/// A group is only ever ticked by a request naming it, so a group whose consumers all died is +/// never reclaimed on its own: it holds its slot against `max_groups` and its members against +/// `max_total_members` forever. Clients that mint a fresh group id per run, which +/// `kafka-console-consumer` does, then walk the gateway into permanent rejection. This runs only +/// where a cap is about to reject, so the cost lands on the path that would otherwise wedge and +/// never on the hot path. +fn reclaim_expired(groups: &mut Groups, now: Instant, except: &StrBytes) -> usize { + let before = groups.len(); + // `except` is the group the caller is mid-way through admitting. It is legitimately empty + // until its first member lands, so sweeping it here would delete the group out from under + // the request that just created it. + groups.retain(|id, group| { + if id == except { + return true; + } + group.tick(now); + !group.is_empty() + }); + before - groups.len() +} + +/// Expire what is overdue in `group_id`. `false` means the group is gone: either it never +/// existed, or ticking emptied it. +fn tick_group(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> bool { + let Some(group) = groups.get_mut(group_id) else { + return false; + }; + group.tick(now); + if group.is_empty() { + groups.remove(group_id); + return false; + } + true +} + +/// An upper bound on what one member adds to the leader's roster. Only the selected protocol's +/// metadata is sent, and which protocol wins is not known until the join completes, so the +/// largest one stands in for it. +fn roster_entry_bytes( + member_id: &StrBytes, + group_instance_id: Option<&StrBytes>, + protocols: &[(StrBytes, Bytes)], +) -> usize { + member_id.len() + + group_instance_id.map_or(0, |id| id.len()) + + protocols + .iter() + .map(|(_, metadata)| metadata.len()) + .max() + .unwrap_or(0) +} + +/// Everything a `JoinGroup` can be rejected for before any group is touched. +fn join_request_error(config: &GroupCoordinatorConfig, request: &JoinRequest) -> Option<i16> { + if request.group_id.is_empty() || request.group_id.len() > MAX_GROUP_ID_BYTES { + return Some(ERROR_INVALID_GROUP_ID); + } + if request.session_timeout < config.min_session_timeout + || request.session_timeout > config.max_session_timeout + { + return Some(ERROR_INVALID_SESSION_TIMEOUT); + } + if request.protocol_type.is_empty() || request.protocols.is_empty() { Review Comment: join_request_error caps protocol bytes (64 KiB), not count. Guard allows 4096 names. select_protocol / supports (state.rs:422, :446) O(p²·n) under global Mutex. Unauth 1000 members × 4096 names stalls every Join/Heartbeat. Extra RAM ≈ vec slots, not tens of GiB. Stock clients send 1–2 names. **Fix**: cap protocols (≤16) before lock. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
