(iggy) branch fix_leave_cg_error created (now 7aa527b7c)

[gk...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

gkoszyk pushed a change to branch fix_leave_cg_error
in repository https://gitbox.apache.org/repos/asf/iggy.git


      at 7aa527b7c fix(server): validate is client part of consumer_group for leave_consumer_group command

This branch includes the following new commits:

     new 7aa527b7c fix(server): validate is client part of consumer_group for leave_consumer_group command

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(iggy) 01/01: fix(server): validate is client part of consumer_group for leave_consumer_group command

[gk...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

gkoszyk pushed a commit to branch fix_leave_cg_error
in repository https://gitbox.apache.org/repos/asf/iggy.git

commit 7aa527b7c41489808863a89cd977233b04628425
Author: numminex <g....@gmail.com>
AuthorDate: Thu Nov 13 21:15:05 2025 +0100

    fix(server): validate is client part of consumer_group for leave_consumer_group command
---
 core/server/src/shard/system/consumer_groups.rs | 10 ++++++++--
 core/server/src/streaming/topics/helpers.rs     | 12 +++++++-----
 2 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/core/server/src/shard/system/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
index f1b310bce..b5c437acd 100644
--- a/core/server/src/shard/system/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -267,12 +267,18 @@ impl IggyShard {
         group_id: &Identifier,
         client_id: u32,
     ) -> Result<(), IggyError> {
-        self.streams.with_consumer_group_by_id_mut(
+        let Some(_) = self.streams.with_consumer_group_by_id_mut(
             stream_id,
             topic_id,
             group_id,
             topics::helpers::leave_consumer_group(client_id),
-        );
+        ) else {
+            return Err(IggyError::ConsumerGroupMemberNotFound(
+                client_id,
+                group_id.clone(),
+                topic_id.clone(),
+            ));
+        };
 
         // Update ClientManager state
         let stream_id_value = self
diff --git a/core/server/src/streaming/topics/helpers.rs b/core/server/src/streaming/topics/helpers.rs
index ee953bca3..4f976cfe2 100644
--- a/core/server/src/streaming/topics/helpers.rs
+++ b/core/server/src/streaming/topics/helpers.rs
@@ -172,11 +172,13 @@ pub fn join_consumer_group(client_id: u32) -> impl FnOnce(ComponentsById<Consume
     }
 }
 
-pub fn leave_consumer_group(client_id: u32) -> impl FnOnce(ComponentsById<ConsumerGroupRefMut>) {
+pub fn leave_consumer_group(
+    client_id: u32,
+) -> impl FnOnce(ComponentsById<ConsumerGroupRefMut>) -> Option<usize> {
     move |(root, members)| {
         let partitions = root.partitions();
         let id = root.id();
-        delete_member(id, client_id, members, partitions);
+        delete_member(id, client_id, members, partitions)
     }
 }
 
@@ -266,13 +268,12 @@ fn delete_member(
     client_id: u32,
     members: &mut ConsumerGroupMembers,
     partitions: &[usize],
-) {
+) -> Option<usize> {
     let member_id = members
         .inner()
         .shared_get()
         .iter()
-        .find_map(|(_, member)| (member.client_id == client_id).then_some(member.id))
-        .expect("delete_member: find member in consumer group slab");
+        .find_map(|(_, member)| (member.client_id == client_id).then_some(member.id))?;
     members.inner_mut().rcu(|members| {
         let mut members = mimic_members(members);
         members.remove(member_id);
@@ -283,6 +284,7 @@ fn delete_member(
         assign_partitions_to_members(id, &mut members, partitions);
         members
     });
+    Some(member_id)
 }
 
 fn assign_partitions_to_members(id: usize, members: &mut Slab<Member>, partitions: &[usize]) {