You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/09/23 09:22:21 UTC
[PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
numinnex opened a new pull request, #4270:
URL: https://github.com/apache/iggy/pull/4270
(no comment)
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5796340790
> again conflicts please
I will resolve them once the stack beneath this PR gets merged (my previous PRs).
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5806311630
/author
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5796334444
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5792330278
Merge after #4263
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5795370967
/author
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5795373928
again conflicts please
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #4270:
URL: https://github.com/apache/iggy/pull/4270
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4270:
URL: https://github.com/apache/iggy/pull/4270#discussion_r4089108959
##########
gateways/kafka/docs/MANUAL_TESTING.md:
##########
@@ -175,8 +180,13 @@ 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 — still “assigned partitions”. G6 empty. Metadata stub error 3. Fix: match G6.
##########
gateways/kafka/docs/SCOPE.md:
##########
@@ -48,6 +49,11 @@ it knows the server supports flexible encoding.
| 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | Decode request; stub response |
| 2 | ListOffsets | 1 | 6 | 1, 2, 3, 4, 5, 6 | Decode request; stub response |
| 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns `NOT_CONTROLLER` (41); `-1` partitions/RF = broker default on v4+ |
+| 10 | FindCoordinator | 0 | 4 | 0, 1, 2, 3, 4 | Answers "this gateway" for group keys; `INVALID_REQUEST` (42) for transaction/share keys; flexible encoding at v3+ |
Review Comment:
still “42 for transaction/share”. Code txn 53.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -0,0 +1,2732 @@
+// 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, HashSet};
+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, LeaveRequest, LeaveResult,
+ LeavingMember, LeftMember, SyncRequest, SyncResult,
+};
+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,
+};
+
+/// 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,
+ },
+}
+
+/// What one `LeaveGroup` removed, which decides how the group reacts afterwards.
+#[derive(Default)]
+struct Departures {
+ members: bool,
+ pending: bool,
+}
+
+/// Member ids per `group_instance_id`. Without KIP-345 replacement one instance id can be held by
+/// several members, so each maps to a set.
+type InstanceHolders = HashMap<StrBytes, HashSet<StrBytes>>;
+
+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();
+ }
+ }
+
+ fn instance_holders(&self) -> InstanceHolders {
+ let mut holders = InstanceHolders::new();
+ for (member_id, member) in &self.members {
+ if let Some(instance_id) = &member.group_instance_id {
+ holders
+ .entry(instance_id.clone())
+ .or_default()
+ .insert(member_id.clone());
+ }
+ }
+ holders
+ }
+
+ /// Removes one `LeaveGroup` identity, following Kafka's `handleLeaveGroup` order.
+ ///
+ /// `holders` must stay exact across calls: a later identity in the same request is resolved
+ /// against it.
+ fn leave_one(
+ &mut self,
+ identity: &LeavingMember,
+ holders: &mut InstanceHolders,
+ departed: &mut Departures,
+ ) -> i16 {
+ let holders_of = identity
+ .group_instance_id
+ .as_ref()
+ .and_then(|instance_id| holders.get_mut(instance_id))
+ .filter(|ids| !ids.is_empty());
+
+ if identity.member_id.is_empty() {
+ let Some(ids) = holders_of else {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ };
+ for id in std::mem::take(ids) {
+ self.remove_member(&id);
+ }
+ departed.members = true;
+ return ERROR_NONE;
+ }
+ if self.pending.remove(&identity.member_id).is_some() {
+ departed.pending = true;
+ return ERROR_NONE;
+ }
+ match (identity.group_instance_id.is_some(), holders_of) {
+ (_, Some(ids)) if ids.contains(&identity.member_id) => {}
+ (_, Some(_)) => return ERROR_FENCED_INSTANCE_ID,
+ (true, None) => return ERROR_UNKNOWN_MEMBER_ID,
+ (false, None) if !self.members.contains_key(&identity.member_id) => {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ (false, None) => {}
+ }
+ if let Some(ids) = self
+ .members
+ .get(&identity.member_id)
+ .and_then(|member| member.group_instance_id.as_ref())
+ .and_then(|instance_id| holders.get_mut(instance_id))
+ {
+ ids.remove(&identity.member_id);
+ }
+ self.remove_member(&identity.member_id);
+ departed.members = true;
+ ERROR_NONE
+ }
+
+ /// React to a `LeaveGroup` the way the session sweep reacts to an expiry.
+ ///
+ /// The rebalance opens only after the removals, so its join window is sized from the
+ /// members that remain. The final `bump` is unconditional on any removal: it is what answers
+ /// a waiter the departed member still has parked on another connection.
+ fn after_departure(&mut self, departed: &Departures, now: Instant) {
+ if departed.members && !self.members.is_empty() && self.phase != Phase::PreparingRebalance {
+ self.prepare_rebalance(now, None);
+ }
+ // A join window left open with nobody in it would close on a later tick and clear the
+ // ids still on their way back. The next admitted member reopens the barrier.
+ if self.members.is_empty() && !self.pending.is_empty() {
+ self.join_deadline = None;
+ }
+ if self.phase == Phase::PreparingRebalance {
+ self.maybe_complete_join(now);
+ }
+ if departed.members || departed.pending {
+ self.bump();
+ }
+ }
+
+ /// 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());
+ 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
+}
+
+/// 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() {
+ return Some(ERROR_INCONSISTENT_GROUP_PROTOCOL);
+ }
+ // Names count too: they are retained alongside the metadata, and a request can carry many
+ // long ones while declaring almost no metadata at all.
+ let retained_bytes: usize = request
+ .protocols
+ .iter()
+ .map(|(name, metadata)| name.len() + metadata.len())
+ .sum::<usize>()
+ + request
+ .group_instance_id
+ .as_ref()
+ .map_or(0, |id| id.as_str().len());
+ if retained_bytes > config.max_member_blob_bytes {
+ return Some(ERROR_INVALID_REQUEST);
+ }
+ None
+}
+
+/// 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
+/// request would multiply the rebalance wakeup storm by the size of the whole map, while a group
+/// that emptied only matters at the moment its slot is needed.
+fn ensure_group(
+ groups: &mut Groups,
+ config: &GroupCoordinatorConfig,
+ request: &JoinRequest,
+ now: Instant,
+) -> Result<bool, i16> {
+ if groups.contains_key(&request.group_id) && tick_group(groups, &request.group_id, now) {
+ return Ok(false);
+ }
+ if !request.member_id.is_empty() {
+ return Err(ERROR_UNKNOWN_MEMBER_ID);
+ }
+ if groups.len() >= config.max_groups {
+ reclaim_expired(groups, now, &request.group_id);
+ }
+ if groups.len() >= config.max_groups {
+ tracing::warn!(
+ max_groups = config.max_groups,
+ "consumer group limit reached; rejecting JoinGroup"
+ );
+ return Err(ERROR_COORDINATOR_NOT_AVAILABLE);
+ }
+ groups.insert(
+ request.group_id.clone(),
+ GroupState::new(request.protocol_type.clone(), now),
+ );
+ Ok(true)
+}
+
+pub fn join_step(
+ groups: &mut Groups,
+ config: &GroupCoordinatorConfig,
+ request: &JoinRequest,
+ now: Instant,
+) -> Step<JoinResult> {
+ let reject = |error: i16| Step::Respond(JoinResult::error(error, request.member_id.clone()));
+ // A group this call created must not outlive a rejection further down: the checks below run
+ // after the insert, and a group left behind holds its `max_groups` slot with no member to
+ // ever expire and release it.
+ let reject_created = |groups: &mut Groups, created: bool, error: i16| {
+ if created {
+ groups.remove(&request.group_id);
+ }
+ Step::Respond(JoinResult::error(error, request.member_id.clone()))
+ };
+
+ if let Some(error) = join_request_error(config, request) {
+ return reject(error);
+ }
+
+ let created = match ensure_group(groups, config, request, now) {
+ Ok(fresh) => fresh,
+ Err(error) => return reject(error),
+ };
+
+ if request.member_id.is_empty()
+ && groups
+ .values()
+ .map(|group| group.members.len() + group.pending.len())
+ .sum::<usize>()
+ >= config.max_total_members
+ {
+ reclaim_expired(groups, now, &request.group_id);
+ }
+ let at_capacity = request.member_id.is_empty()
+ && groups
+ .values()
+ .map(|group| group.members.len() + group.pending.len())
+ .sum::<usize>()
+ >= config.max_total_members;
+
+ let Some(group) = groups.get_mut(&request.group_id) else {
+ return reject(ERROR_UNKNOWN_MEMBER_ID);
+ };
+
+ if group.protocol_type != request.protocol_type
+ || !group.protocols_compatible(&request.member_id, &request.protocols)
+ {
+ return reject_created(groups, created, ERROR_INCONSISTENT_GROUP_PROTOCOL);
+ }
+
+ if request.member_id.is_empty() {
+ // Decided before touching `group` again so the borrow ends before the cleanup below.
+ let capacity_error =
+ if group.members.len() + group.pending.len() >= config.max_members_per_group {
+ Some(ERROR_GROUP_MAX_SIZE_REACHED)
+ } else if at_capacity {
+ tracing::warn!(
+ max_total_members = config.max_total_members,
+ "consumer group member limit reached; rejecting JoinGroup"
+ );
+ Some(ERROR_COORDINATOR_NOT_AVAILABLE)
+ } else {
+ None
+ };
+ if let Some(error) = capacity_error {
+ return reject_created(groups, created, error);
+ }
+ let Some(group) = groups.get_mut(&request.group_id) else {
+ return reject(ERROR_UNKNOWN_MEMBER_ID);
+ };
+ let member_id = generate_member_id(request.group_instance_id.as_ref());
+ if request.require_known_member_id {
+ group
+ .pending
+ .insert(member_id.clone(), now + request.session_timeout);
+ group.bump();
+ return Step::Respond(JoinResult::error(ERROR_MEMBER_ID_REQUIRED, member_id));
+ }
+ return admit(group, config, &member_id, request, now);
+ }
+
+ if group.pending.contains_key(&request.member_id) {
+ return admit(group, config, &request.member_id, request, now);
+ }
+ let Some(member) = group.members.get(&request.member_id) else {
+ return reject(ERROR_UNKNOWN_MEMBER_ID);
+ };
+ let protocols_changed = member.protocols != request.protocols;
+ let is_leader = group.leader.as_ref() == Some(&request.member_id);
+
+ rejoin_step(group, config, request, is_leader, protocols_changed, now)
+}
+
+/// Dispatch for a member the group already knows, by phase.
+fn rejoin_step(
+ group: &mut GroupState,
+ config: &GroupCoordinatorConfig,
+ request: &JoinRequest,
+ is_leader: bool,
+ protocols_changed: bool,
+ now: Instant,
+) -> Step<JoinResult> {
+ match group.phase {
+ Phase::PreparingRebalance => {
+ if let Some(member) = group.members.get_mut(&request.member_id) {
+ member.rejoin(request, now, config.max_rebalance_timeout);
+ }
+ group.bump();
+ group.maybe_complete_join(now);
+ park_or_respond(group, &request.member_id, now)
+ }
+ Phase::Stable if is_leader || protocols_changed => {
+ if let Some(member) = group.members.get_mut(&request.member_id) {
+ member.rejoin(request, now, config.max_rebalance_timeout);
+ }
+ group.prepare_rebalance(now, Some(&request.member_id));
+ group.maybe_complete_join(now);
+ park_or_respond(group, &request.member_id, now)
+ }
+ Phase::CompletingRebalance if protocols_changed => {
+ if let Some(member) = group.members.get_mut(&request.member_id) {
+ member.rejoin(request, now, config.max_rebalance_timeout);
+ }
+ group.prepare_rebalance(now, Some(&request.member_id));
+ group.maybe_complete_join(now);
+ park_or_respond(group, &request.member_id, now)
+ }
+ Phase::Stable | Phase::CompletingRebalance => {
+ if let Some(member) = group.members.get_mut(&request.member_id) {
+ member.session_timeout = request.session_timeout;
+ member.session_deadline = now + request.session_timeout;
+ // Answering without consuming the snapshot would leave an uncollected one on a
+ // member that is not parked, which a later rejoin would serve as a stale answer.
+ member.join_response = None;
+ }
+ Step::Respond(group.current_generation_result(&request.member_id))
+ }
+ }
+}
+
+/// Keep a member parked on a barrier alive across the session sweep.
+///
+/// A parked member is inside its own `JoinGroup` or `SyncGroup` call and cannot send a
+/// heartbeat, so
+/// the sweep would evict the member that did the right thing while one that never rejoined
+/// survives on its heartbeats. Kafka exempts them outright
+/// (`MemberMetadata.hasSatisfiedHeartbeat`: `isAwaitingJoin || isAwaitingSync`). Refreshing
+/// rather than exempting keeps the member's own session deadline chained in `next_deadline`,
+/// which is what stops a park from pinning a connection for the whole rebalance window.
+fn refresh_parked_session(
+ groups: &mut Groups,
+ group_id: &StrBytes,
+ member_id: &StrBytes,
+ now: Instant,
+) {
+ if let Some(group) = groups.get_mut(group_id)
+ && let Some(member) = group.members.get_mut(member_id)
+ {
+ member.session_deadline = now + member.session_timeout;
+ }
+}
+
+pub fn join_resume_step(
+ groups: &mut Groups,
+ group_id: &StrBytes,
+ member_id: &StrBytes,
+ now: Instant,
+) -> Step<JoinResult> {
+ refresh_parked_session(groups, group_id, member_id, now);
+ if !tick_group(groups, group_id, now) {
+ return Step::Respond(JoinResult::error(
+ ERROR_UNKNOWN_MEMBER_ID,
+ member_id.clone(),
+ ));
+ }
+ let Some(group) = groups.get_mut(group_id) else {
+ return Step::Respond(JoinResult::error(
+ ERROR_UNKNOWN_MEMBER_ID,
+ member_id.clone(),
+ ));
+ };
+ park_or_respond(group, member_id, now)
+}
+
+pub fn heartbeat_step(
+ groups: &mut Groups,
+ group_id: &StrBytes,
+ generation_id: i32,
+ member_id: &StrBytes,
+ now: Instant,
+) -> i16 {
+ if !tick_group(groups, group_id, now) {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ let Some(group) = groups.get_mut(group_id) else {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ };
+ if !group.members.contains_key(member_id) {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ if generation_id != group.generation_id {
+ return ERROR_ILLEGAL_GENERATION;
+ }
+ if let Some(member) = group.members.get_mut(member_id) {
+ member.session_deadline = now + member.session_timeout;
+ }
+ if group.phase == Phase::PreparingRebalance {
+ ERROR_REBALANCE_IN_PROGRESS
+ } else {
+ ERROR_NONE
+ }
+}
+
+pub fn sync_step(
+ groups: &mut Groups,
+ config: &GroupCoordinatorConfig,
+ request: &SyncRequest,
+ now: Instant,
+) -> Step<SyncResult> {
+ if !tick_group(groups, &request.group_id, now) {
+ return Step::Respond(SyncResult::error(ERROR_UNKNOWN_MEMBER_ID));
+ }
+ let oversized = request
+ .assignments
+ .iter()
+ .any(|(_, blob)| blob.len() > config.max_member_blob_bytes);
+
+ let Some(group) = groups.get_mut(&request.group_id) else {
+ return Step::Respond(SyncResult::error(ERROR_UNKNOWN_MEMBER_ID));
+ };
+ if !group.members.contains_key(&request.member_id) {
+ return Step::Respond(SyncResult::error(ERROR_UNKNOWN_MEMBER_ID));
+ }
+ if request.generation_id != group.generation_id {
+ return Step::Respond(SyncResult::error(ERROR_ILLEGAL_GENERATION));
+ }
+ let protocol_mismatch = request
+ .protocol_type
+ .as_ref()
+ .is_some_and(|protocol_type| *protocol_type != group.protocol_type)
+ || request
+ .protocol_name
+ .as_ref()
+ .is_some_and(|name| Some(name) != group.protocol_name.as_ref());
+ if protocol_mismatch {
+ return Step::Respond(SyncResult::error(ERROR_INCONSISTENT_GROUP_PROTOCOL));
+ }
+ if oversized {
+ return Step::Respond(SyncResult::error(ERROR_INVALID_REQUEST));
+ }
+
+ match group.phase {
+ Phase::PreparingRebalance => Step::Respond(SyncResult::error(ERROR_REBALANCE_IN_PROGRESS)),
+ Phase::Stable => {
+ if let Some(member) = group.members.get_mut(&request.member_id) {
+ member.session_deadline = now + member.session_timeout;
+ }
+ Step::Respond(group.sync_result(&request.member_id))
+ }
+ Phase::CompletingRebalance => {
+ if group.leader.as_ref() == Some(&request.member_id) {
+ group.apply_assignments(&request.assignments, now);
+ return Step::Respond(group.sync_result(&request.member_id));
+ }
+ if let Some(member) = group.members.get_mut(&request.member_id) {
+ member.synced = true;
+ member.session_deadline = now + member.session_timeout;
+ }
+ Step::Wait {
+ member_id: request.member_id.clone(),
+ wake_at: group.wake_at(now),
+ }
+ }
+ }
+}
+
+pub fn sync_resume_step(
+ groups: &mut Groups,
+ group_id: &StrBytes,
+ member_id: &StrBytes,
+ generation_id: i32,
+ now: Instant,
+) -> Step<SyncResult> {
+ refresh_parked_session(groups, group_id, member_id, now);
+ if !tick_group(groups, group_id, now) {
+ return Step::Respond(SyncResult::error(ERROR_UNKNOWN_MEMBER_ID));
+ }
+ let Some(group) = groups.get_mut(group_id) else {
+ return Step::Respond(SyncResult::error(ERROR_UNKNOWN_MEMBER_ID));
+ };
+ if !group.members.contains_key(member_id) {
+ return Step::Respond(SyncResult::error(ERROR_UNKNOWN_MEMBER_ID));
+ }
+ if group.generation_id != generation_id || group.phase == Phase::PreparingRebalance {
+ return Step::Respond(SyncResult::error(ERROR_REBALANCE_IN_PROGRESS));
+ }
+ if group.phase == Phase::Stable {
+ return Step::Respond(group.sync_result(member_id));
+ }
+ Step::Wait {
+ member_id: member_id.clone(),
+ wake_at: group.wake_at(now),
+ }
+}
+
+/// Never creates a group, and removes one the leave empties rather than leaving it for
+/// `reclaim_expired`.
+pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) -> LeaveResult {
+ if request.group_id.is_empty() {
+ return LeaveResult::error(ERROR_INVALID_GROUP_ID);
+ }
+ let unknown = || LeaveResult {
+ error: ERROR_NONE,
+ members: request
+ .members
+ .iter()
+ .map(|identity| LeftMember::from((identity, ERROR_UNKNOWN_MEMBER_ID)))
+ .collect(),
+ };
+ if !tick_group(groups, &request.group_id, now) {
Review Comment:
leave_step tick_group first. now >= join_deadline + last member still in map → complete_join pending.clear() (:446) before after_departure cancel (:392). Same wipe: tick session/window (:282). New test leaves at now, window still open. **Fix**: skip complete_join when pending nonempty and members would empty.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5809884631
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4270:
URL: https://github.com/apache/iggy/pull/4270#discussion_r4082646198
##########
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(
+ version,
+ &keys,
+ ERROR_INVALID_REQUEST,
Review Comment:
key_type!=0 → 42. librdkafka rd_kafka_txn_handle_FindCoordinator default → txn_coord_set(NULL) → 500ms forever. Java fatal. File not in 4270-vs-4263 delta; PR vs master still ships. Fix: 53 or 31. (same as in #4263)
##########
gateways/kafka/docs/MANUAL_TESTING.md:
##########
@@ -175,8 +180,13 @@ 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 |
+| G4 | Ungraceful consumer exit | `kill -9` one of G3's kcats | Within `session.timeout.ms` the survivor logs a rebalance and is assigned every partition |
+| G5 | Java console consumer | `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g2 --topic test` | Exercises JoinGroup v9, SyncGroup v5, Heartbeat v4, FindCoordinator v4. A 4.0 client may need `--consumer-property group.protocol=classic`, or it sends ConsumerGroupHeartbeat (68) and the connection closes |
+| G6 | Graceful kcat exit | G3's two kcats, then Ctrl-C one (kcat closes its consumer, which sends LeaveGroup v0/v1) | The survivor rebalances within one heartbeat interval, not `session.timeout.ms`, and is assigned every partition |
Review Comment:
G6/G7 — “assigned partitions”. Metadata stub → RangeAssignor 0. OffsetFetch honesty OK. **Fix**: join/leave/sync real; assignment empty.
##########
gateways/kafka/docs/kafka_api_keys_reference.md:
##########
Review Comment:
“4.0 default key 68” false. group.protocol default classic.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -0,0 +1,2698 @@
+// 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, HashSet};
+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, LeaveRequest, LeaveResult,
+ LeavingMember, LeftMember, SyncRequest, SyncResult,
+};
+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,
+};
+
+/// 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,
+ },
+}
+
+/// What one `LeaveGroup` removed, which decides how the group reacts afterwards.
+#[derive(Default)]
+struct Departures {
+ members: bool,
+ pending: bool,
+}
+
+/// Member ids per `group_instance_id`. Without KIP-345 replacement one instance id can be held by
+/// several members, so each maps to a set.
+type InstanceHolders = HashMap<StrBytes, HashSet<StrBytes>>;
+
+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();
+ }
+ }
+
+ fn instance_holders(&self) -> InstanceHolders {
+ let mut holders = InstanceHolders::new();
+ for (member_id, member) in &self.members {
+ if let Some(instance_id) = &member.group_instance_id {
+ holders
+ .entry(instance_id.clone())
+ .or_default()
+ .insert(member_id.clone());
+ }
+ }
+ holders
+ }
+
+ /// Removes one `LeaveGroup` identity, following Kafka's `handleLeaveGroup` order.
+ ///
+ /// `holders` must stay exact across calls: a later identity in the same request is resolved
+ /// against it.
+ fn leave_one(
+ &mut self,
+ identity: &LeavingMember,
+ holders: &mut InstanceHolders,
+ departed: &mut Departures,
+ ) -> i16 {
+ let holders_of = identity
+ .group_instance_id
+ .as_ref()
+ .and_then(|instance_id| holders.get_mut(instance_id))
+ .filter(|ids| !ids.is_empty());
+
+ if identity.member_id.is_empty() {
+ let Some(ids) = holders_of else {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ };
+ for id in std::mem::take(ids) {
+ self.remove_member(&id);
+ }
+ departed.members = true;
+ return ERROR_NONE;
+ }
+ if self.pending.remove(&identity.member_id).is_some() {
+ departed.pending = true;
+ return ERROR_NONE;
+ }
+ match (identity.group_instance_id.is_some(), holders_of) {
+ (_, Some(ids)) if ids.contains(&identity.member_id) => {}
+ (_, Some(_)) => return ERROR_FENCED_INSTANCE_ID,
+ (true, None) => return ERROR_UNKNOWN_MEMBER_ID,
+ (false, None) if !self.members.contains_key(&identity.member_id) => {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ (false, None) => {}
+ }
+ if let Some(ids) = self
+ .members
+ .get(&identity.member_id)
+ .and_then(|member| member.group_instance_id.as_ref())
+ .and_then(|instance_id| holders.get_mut(instance_id))
+ {
+ ids.remove(&identity.member_id);
+ }
+ self.remove_member(&identity.member_id);
+ departed.members = true;
+ ERROR_NONE
+ }
+
+ /// React to a `LeaveGroup` the way the session sweep reacts to an expiry.
+ ///
+ /// The rebalance opens only after the removals, so its join window is sized from the
+ /// members that remain. The final `bump` is unconditional on any removal: it is what answers
+ /// a waiter the departed member still has parked on another connection.
+ fn after_departure(&mut self, departed: &Departures, now: Instant) {
+ if departed.members && !self.members.is_empty() && self.phase != Phase::PreparingRebalance {
Review Comment:
last-member leave while PreparingRebalance skips prepare_rebalance; join_deadline stays. Later tick → complete_join pending.clear() (:441) → group drop. Pending rejoin 25. Kafka initNextGeneration keeps pendingJoinMembers. Test :2425 same Instant only. KIP-394 empty rejoin recovers; no durable loss. **Fix**: join_deadline=None when members.empty && pending nonempty.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): handle Kafka LeaveGroup for consumer groups [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4270:
URL: https://github.com/apache/iggy/pull/4270#issuecomment-5832926588
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4270?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 26.02%. Comparing base ([`5d8129e`](https://app.codecov.io/gh/apache/iggy/commit/5d8129e95de1d1fdf2ed42af015340b82af1c59a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`57853b8`](https://app.codecov.io/gh/apache/iggy/commit/57853b8eee53e6d4c0ce3b0cb706cda96e04a098?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4270 +/- ##
=============================================
- Coverage 87.66% 26.02% -61.65%
+ Complexity 1576 1575 -1
=============================================
Files 1284 1281 -3
Lines 225457 181922 -43535
Branches 188821 145288 -43533
=============================================
- Hits 197657 47352 -150305
- Misses 23080 132912 +109832
+ Partials 4720 1658 -3062
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `11.75% <ø> (-77.02%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.68% <ø> (-0.02%)` | :arrow_down: |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.25% <ø> (-19.17%)` | :arrow_down: |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.97% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.67% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.43% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4270/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.11% <ø> (+0.05%)` | :arrow_up: |
[see 744 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4270/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org