ryerraguntla commented on code in PR #4263: URL: https://github.com/apache/iggy/pull/4263#discussion_r4082513427
########## gateways/kafka/docs/CONSUMER_GROUPS.md: ########## @@ -0,0 +1,123 @@ +# Consumer group coordination + +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/`. 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 | +| --- | --- | --- | --- | +| 10 | FindCoordinator | 0-4 | Always answers "this gateway", the same node the Metadata broker list advertises | +| 11 | JoinGroup | 0-9 | Parks until the group's join barrier completes | +| 12 | Heartbeat | 0-4 | Refreshes a session; `REBALANCE_IN_PROGRESS` is how a follower learns to rejoin | +| 14 | SyncGroup | 0-5 | Relays the leader's assignment blobs; a follower parks until the leader syncs | + +`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. + +## Assignment is the client's job + +The gateway elects a leader, hands it every member's subscription metadata, and fans the blobs the +leader returns back out - one per member, exactly the bytes filed under that member's id, in the +same generation. It never decodes `ConsumerProtocolSubscription` or `ConsumerProtocolAssignment`. + +The practical consequence: **whatever assignor the client ships is the assignor** - range, +round-robin, sticky, cooperative-sticky, or a custom one. There is no supported subset to +document, and nothing to configure. Partitions come out disjoint because the leader's assignor +made them disjoint. + +When two members list different protocols, the coordinator picks the one every member supports +with the most first-preference votes; a tie is broken by the leader's own list order. Kafka breaks +that tie by set iteration order, which is not deterministic - this one is, deliberately, so tests +can pin it. + +## One gateway per bootstrap endpoint + +Group membership is process memory. Two gateway instances fronting one Iggy cluster do **not** +share it, and a client that can reach both ends up in two independent groups under one name: + +1. Each gateway advertises itself as the coordinator. A client that joined on gateway A and later + connects to B is told `UNKNOWN_MEMBER_ID` and rejoins on B with a fresh id. +2. Each group's leader assigns **all** partitions to its own members, so every partition has two + consumers and every record is delivered twice. +3. Both groups commit against the same Iggy key, so committed offsets flip between the two + groups' positions. + +The rule that follows: one gateway per Kafka bootstrap endpoint, with +`IGGY_KAFKA_ADVERTISED_HOST`/`_PORT` routing back to that same instance, and no load balancer in +front of more than one gateway for consumers. Fixing this needs a shared coordinator (Iggy-backed +group state, or gateway-to-gateway forwarding) and is out of scope here. + +Group state also does not survive a gateway restart. Clients recover on their own - a member id +the coordinator does not know is answered with `UNKNOWN_MEMBER_ID`, which is what makes the Java +and librdkafka clients discard it and join from scratch - and committed offsets are unaffected +because they live in Iggy. + +## Timeouts without a background task + +There is no timer thread. Every request that touches a group first expires whatever is overdue in +it (sessions, unclaimed member ids, an elapsed join or sync window), and a parked `JoinGroup` or +`SyncGroup` handler sleeps until that group's next deadline. The coroutine waiting on a barrier is +therefore also the timer that fires it. + +Expiry is judged against the recorded deadline, not against when the sweep happens to run, so a +heartbeat that arrives after its own member's deadline finds the member already gone. A broker +whose timer thread has not fired yet would still accept it. This is stricter, and deterministic. + +What this does not do is evict a member of a group **nobody is talking to**. If the surviving +members are heartbeating, a dead member is evicted at most one heartbeat interval after its +deadline, and that same heartbeat returns `REBALANCE_IN_PROGRESS`, so the rebalance starts in the +same round trip. If *every* member is dead, nobody is evicted until the next request for that +group, at which point the joiner clears the stale ids and completes immediately. No consumer can +observe the difference, because there is no consumer. The only cost is stale memory, bounded by +the caps below. + +## Static membership is accepted, not honoured + +`group.instance.id` (JoinGroup v5+) is stored and echoed back to the leader, and it seeds the +generated member id so a static member is recognisable in logs. Nothing else about KIP-345 is +implemented: there is no `FENCED_INSTANCE_ID`, and a returning static member is **not** matched to +its previous identity - it is a new dynamic member and its rejoin triggers a rebalance like any +other. Full static membership belongs to +[#3543](https://github.com/apache/iggy/issues/3543). + +## Capacity caps + +`GroupCoordinatorConfig` (no environment variables yet; `GatewayConfig.group`): + +| Setting | Default | Meaning | +| --- | --- | --- | +| `min_session_timeout` / `max_session_timeout` | 6s / 30min | Kafka's `group.min/max.session.timeout.ms`; outside the range is `INVALID_SESSION_TIMEOUT` (26) | +| `initial_rebalance_delay` | 3s | Kafka's `group.initial.rebalance.delay.ms`; a new group waits this out so consumers starting together land in one generation | +| `max_groups` | 1000 | Beyond it, a new group is `COORDINATOR_NOT_AVAILABLE` (15, retriable) | +| `max_members_per_group` | 1000 | Kafka's `group.max.size`; beyond it, `GROUP_MAX_SIZE_REACHED` (81) | +| `max_total_members` | 10000 | Across every group, checked before a member id is minted; `COORDINATOR_NOT_AVAILABLE` (15) | +| `max_member_blob_bytes` | 64 KiB | One JoinGroup's total protocol metadata, and one SyncGroup assignment blob; beyond it, `INVALID_REQUEST` (42) | + +The frame-level bounds guard (`src/protocol/bounds_guard.rs`) bounds one request. These bound what +is *retained*: a member's subscription and assignment outlive the connection that sent them, up to +`max_session_timeout`. Worst case at the defaults is roughly 1.3 GiB of opaque bytes. + +A group id is capped at 246 bytes, not Kafka's 249: the Iggy offset key is `kafka.cg.<group>` and +an Iggy name caps at 255. A longer id is `INVALID_GROUP_ID` (24) here rather than a failure later +at commit time. + +## What a real consumer still cannot do + +`OffsetFetch` (9) is sent by every consumer immediately after `SyncGroup`, and it is not in scope Review Comment: “OffsetFetch after every Sync then close-loop” false. Empty assignment skips OffsetFetch (ConsumerCoordinator.java:967). Stall = empty assignment, not OffsetFetch. **Fix**: rewrite that sentence. ########## gateways/kafka/src/protocol/handlers/find_coordinator.rs: ########## @@ -0,0 +1,164 @@ +// 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. + +//! `FindCoordinator` (API key 10). +//! +//! The answer is always this gateway, matching the single broker entry Metadata advertises. No +//! group state is consulted: where the coordinator lives does not depend on which group is asked +//! about. + +use bytes::Bytes; +use kafka_protocol::messages::find_coordinator_response::Coordinator; +use kafka_protocol::messages::{BrokerId, FindCoordinatorRequest, FindCoordinatorResponse}; +use kafka_protocol::protocol::StrBytes; + +use crate::error::Result; +use crate::protocol::api::{ + API_KEY_FIND_COORDINATOR, ApiVersionRange, BrokerAdvertise, ERROR_INVALID_REQUEST, ERROR_NONE, + ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, is_supported_version, +}; +use crate::protocol::bounds_guard::validate_find_coordinator_shape; +use crate::protocol::handlers::{ + decode_guarded, encode_message, respond_or_close, unsupported_version_response, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_FIND_COORDINATOR, + min_version: 0, + max_version: 4, +}; + +/// `key_type` 0. Types 1 (transaction) and 2 (share) have no coordinator here. +const COORDINATOR_TYPE_GROUP: i8 = 0; + +/// The node id this gateway advertises for itself, in Metadata and here alike. +const SELF_NODE_ID: i32 = 1; + +const UNSUPPORTED_KEY_TYPE_MESSAGE: &str = "only group coordination is supported"; + +#[expect( + clippy::unused_async, + reason = "the shared handler signature, kept until a handler awaits the bridge" +)] +pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { + if !is_supported_version(API_KEY_FIND_COORDINATOR, api_version) { + return unsupported_version_response(API_KEY_FIND_COORDINATOR, api_version, |version| { + encode_error_response(version, ERROR_UNSUPPORTED_VERSION) + }); + } + match decode_guarded::<FindCoordinatorRequest>(api_version, body, |version, body| { + validate_find_coordinator_shape(version, body, state.max_frame_size) + }) { + Ok(request) => respond_or_close( + encode_response(api_version, &request, &state.broker), + "FindCoordinator", + ), + Err(error) => { + // debug!, not warn!: attacker-controlled, not operator-actionable. + tracing::debug!(%error, api_version, "Failed to decode FindCoordinator request"); + if api_version >= 4 { + // v4 carries its error per requested key and the keys are exactly what failed to + // decode, so there is no honest body to send. + return HandleOutcome::Close; + } + respond_or_close( + encode_error_response(api_version, ERROR_INVALID_REQUEST), + "FindCoordinator", + ) + } + } +} + +/// # Errors +/// +/// Returns an error when `kafka_protocol` cannot encode the response at `version`. +pub fn encode_response( + version: i16, + request: &FindCoordinatorRequest, + broker: &BrokerAdvertise, +) -> Result<Bytes> { + let keys = if version >= 4 { + request.coordinator_keys.clone() + } else { + vec![request.key.clone()] + }; + if request.key_type == COORDINATOR_TYPE_GROUP { + encode_inner(version, &keys, ERROR_NONE, None, Some(broker)) + } else { + // Not a retriable code: transactions are out of scope for good, and + // COORDINATOR_NOT_AVAILABLE would make a transactional producer retry forever. + encode_inner( Review Comment: Advertise key 10 + key_type!=0 → error 42. Java txn fatal. librdkafka rd_kafka_txn_handle_FindCoordinator default → txn_coord_set(NULL) → 500ms retry forever. Fatal only 53/31. Pre-PR: missing 10 → __UNSUPPORTED_FEATURE fatal. Regression. **Fix**: return 53 or 31. Do not use 35 (same livelock). ########## gateways/kafka/docs/MANUAL_TESTING.md: ########## @@ -175,8 +179,10 @@ Requires `kcat` installed. Gateway does **not** implement SASL or full broker se | ID | Test | Command | Expected (foundation) | | ---- | ------ | --------- | --------------------- | | G1 | Broker metadata | `kcat -b 127.0.0.1:9093 -L` | ApiVersions + Metadata handshake; broker appears in metadata | -| G2 | Produce (likely fails later) | `echo "hello" \| kcat -b 127.0.0.1:9093 -t test -P` | May fail at coordinator/group stage — document actual error | -| G3 | Consumer (likely fails later) | `kcat -b 127.0.0.1:9093 -t test -C -o beginning` | May fail without consumer groups — document actual error | +| G2 | Produce (likely fails later) | `echo "hello" \| kcat -b 127.0.0.1:9093 -t test -P` | Produce is still a stub: retriable `NOT_LEADER_OR_FOLLOWER` (6), so kcat retries — document actual error | +| G3 | Consumer group rebalance | `kcat -b 127.0.0.1:9093 -G g1 test` in two terminals | Each prints its assigned partitions and the two sets are disjoint; then both stall, because OffsetFetch (9) is unlisted and closes the connection — the client re-runs FindCoordinator and loops. Record the exact librdkafka log lines | Review Comment: G3/G4 claim assigned/disjoint partitions. Metadata stub → RangeAssignor empty. **Fix**: expect join/sync + 0 partitions. Checke Readme.md for the same. ########## gateways/kafka/src/group/state.rs: ########## @@ -0,0 +1,1923 @@ +// 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, +}; +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)| { + ( + StrBytes::from_string(name.as_str().to_owned()), + 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(|id| StrBytes::from_string(id.as_str().to_owned())), + 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(|id| StrBytes::from_string(id.as_str().to_owned())); + 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 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() + } + + 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)| blob.clone()); Review Comment: apply_assignments blob.clone() pins Sync frame. Join already copy_from_slice (state.rs:107). **Fix**: own copy. -- 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]
