You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/11/15 14:42:42 UTC
(iggy) 01/01: fix(server): fix store_consumer_offset for consumer groups
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch fix_consumer_group_offset_commit
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b8a0dcbdd2e13189ac00ae7dafe9b9870ceaf3cd
Author: numminex <g....@gmail.com>
AuthorDate: Sat Nov 15 15:42:30 2025 +0100
fix(server): fix store_consumer_offset for consumer groups
---
core/server/src/bootstrap.rs | 20 ++--
core/server/src/shard/system/consumer_offsets.rs | 46 +++++----
core/server/src/slab/streams.rs | 78 +++++++++++----
.../src/streaming/partitions/consumer_offset.rs | 11 ++-
core/server/src/streaming/partitions/helpers.rs | 36 ++++---
core/server/src/streaming/partitions/partition.rs | 9 +-
core/server/src/streaming/partitions/storage.rs | 110 ++++++++++++++++++---
core/server/src/streaming/polling_consumer.rs | 28 +++++-
core/server/src/streaming/streams/helpers.rs | 4 +-
9 files changed, 258 insertions(+), 84 deletions(-)
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index b5ba07fce..8f8d56dc2 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -43,12 +43,16 @@ use crate::{
state::system::{StreamState, TopicState, UserState},
streaming::{
partitions::{
- consumer_offset::ConsumerOffset, helpers::create_message_deduplicator,
- journal::MemoryMessageJournal, log::SegmentedLog, partition,
- storage::load_consumer_offsets,
+ consumer_offset::ConsumerOffset,
+ helpers::create_message_deduplicator,
+ journal::MemoryMessageJournal,
+ log::SegmentedLog,
+ partition,
+ storage::{load_consumer_group_offsets, load_consumer_offsets},
},
persistence::persister::{FilePersister, FileWithSyncPersister, PersisterKind},
personal_access_tokens::personal_access_token::PersonalAccessToken,
+ polling_consumer::{ConsumerGroupId, MemberId},
segments::{Segment, storage::Storage},
stats::{PartitionStats, StreamStats, TopicStats},
storage::SystemStorage,
@@ -63,7 +67,7 @@ use ahash::HashMap;
use compio::{fs::create_dir_all, runtime::Runtime};
use err_trail::ErrContext;
use iggy_common::{
- ConsumerKind, IggyByteSize, IggyError,
+ IggyByteSize, IggyError,
defaults::{
DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH,
MIN_USERNAME_LENGTH,
@@ -690,7 +694,7 @@ async fn load_partition(
config.get_consumer_group_offsets_path(stream_id, topic_id, partition_id as usize);
let consumer_offset = Arc::new(
- load_consumer_offsets(&consumer_offset_path, ConsumerKind::Consumer)?
+ load_consumer_offsets(&consumer_offset_path)?
.into_iter()
.map(|offset| (offset.consumer_id as usize, offset))
.collect::<HashMap<usize, ConsumerOffset>>()
@@ -698,10 +702,10 @@ async fn load_partition(
);
let consumer_group_offset = Arc::new(
- load_consumer_offsets(&consumer_group_offsets_path, ConsumerKind::ConsumerGroup)?
+ load_consumer_group_offsets(&consumer_group_offsets_path)?
.into_iter()
- .map(|offset| (offset.consumer_id as usize, offset))
- .collect::<HashMap<usize, ConsumerOffset>>()
+ .map(|(consumer_group_id, member_id, offset)| ((consumer_group_id, member_id), offset))
+ .collect::<HashMap<(ConsumerGroupId, MemberId), ConsumerOffset>>()
.into(),
);
diff --git a/core/server/src/shard/system/consumer_offsets.rs b/core/server/src/shard/system/consumer_offsets.rs
index f3f239ff9..dbc1196b2 100644
--- a/core/server/src/shard/system/consumer_offsets.rs
+++ b/core/server/src/shard/system/consumer_offsets.rs
@@ -128,12 +128,17 @@ impl IggyShard {
partition_id,
partitions::helpers::get_consumer_offset(id),
),
- PollingConsumer::ConsumerGroup(_, id) => self.streams.with_partition_by_id(
- stream_id,
- topic_id,
- partition_id,
- partitions::helpers::get_consumer_group_member_offset(id),
- ),
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
+ self.streams.with_partition_by_id(
+ stream_id,
+ topic_id,
+ partition_id,
+ partitions::helpers::get_consumer_group_member_offset(
+ consumer_group_id,
+ member_id,
+ ),
+ )
+ }
};
Ok(offset)
}
@@ -216,13 +221,14 @@ impl IggyShard {
),
);
}
- PollingConsumer::ConsumerGroup(_, id) => {
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
self.streams.with_partition_by_id(
stream_id,
topic_id,
partition_id,
partitions::helpers::store_consumer_group_member_offset(
- *id,
+ *consumer_group_id,
+ *member_id,
stream_id_num,
topic_id_num,
partition_id,
@@ -250,11 +256,11 @@ impl IggyShard {
)
})
}
- PollingConsumer::ConsumerGroup(_, id) => {
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
self.streams
- .with_partition_by_id(stream_id, topic_id, partition_id, partitions::helpers::delete_consumer_group_member_offset(*id)).with_error(|error| {
+ .with_partition_by_id(stream_id, topic_id, partition_id, partitions::helpers::delete_consumer_group_member_offset(*consumer_group_id, *member_id)).with_error(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to delete consumer group member offset for member with ID: {id} in topic with ID: {topic_id} and stream with ID: {stream_id}",
+ "{COMPONENT} (error: {error}) - failed to delete consumer group member offset for member with ID: {member_id:?} in topic with ID: {topic_id} and stream with ID: {stream_id}",
)
})
}
@@ -286,19 +292,21 @@ impl IggyShard {
);
partitions::storage::persist_offset(&path, offset_value).await
}
- PollingConsumer::ConsumerGroup(_, id) => {
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
let (offset_value, path) = self.streams.with_partition_by_id(
stream_id,
topic_id,
partition_id,
- |(.., offsets, _)| {
+ move |(.., offsets, _)| {
let hdl = offsets.pin();
- let item = hdl.get(id).expect(
- "persist_consumer_group_member_offset_to_disk: offset not found",
- );
- let offset = item.offset.load(std::sync::atomic::Ordering::Relaxed);
- let path = item.path.clone();
- (offset, path)
+ let key = (*consumer_group_id, *member_id);
+ let item = hdl
+ .get(&key)
+ .expect("persist_consumer_offset_to_disk: offset not found");
+ (
+ item.offset.load(std::sync::atomic::Ordering::Relaxed),
+ item.path.clone(),
+ )
},
);
partitions::storage::persist_offset(&path, offset_value).await
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 907806f66..6b35ea031 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -17,6 +17,7 @@
use crate::shard::task_registry::TaskRegistry;
use crate::streaming::partitions as streaming_partitions;
+use crate::streaming::partitions::consumer_offset::ConsumerOffset;
use crate::streaming::stats::StreamStats;
use crate::{
binary::handlers::messages::poll_messages_handler::IggyPollMetadata,
@@ -366,27 +367,26 @@ impl MainOps for Streams {
Ok(batches)
}
PollingKind::Next => {
- let (consumer_offset, consumer_id) = match consumer {
- PollingConsumer::Consumer(consumer_id, _) => (
- self.with_partition_by_id(
+ let consumer_offset = match consumer {
+ PollingConsumer::Consumer(consumer_id, _) => self
+ .with_partition_by_id(
stream_id,
topic_id,
partition_id,
streaming_partitions::helpers::get_consumer_offset(consumer_id),
)
.map(|c_offset| c_offset.stored_offset),
- consumer_id,
- ),
- PollingConsumer::ConsumerGroup(cg_id, _) => (
- self.with_partition_by_id(
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => self
+ .with_partition_by_id(
stream_id,
topic_id,
partition_id,
- streaming_partitions::helpers::get_consumer_group_member_offset(cg_id),
+ streaming_partitions::helpers::get_consumer_group_member_offset(
+ consumer_group_id,
+ member_id,
+ ),
)
.map(|cg_offset| cg_offset.stored_offset),
- cg_id,
- ),
};
if consumer_offset.is_none() {
@@ -397,12 +397,25 @@ impl MainOps for Streams {
} else {
let consumer_offset = consumer_offset.unwrap();
let offset = consumer_offset + 1;
- tracing::trace!(
- "Getting next messages for consumer id: {} for partition: {} from offset: {}...",
- consumer_id,
- partition_id,
- offset
- );
+ match consumer {
+ PollingConsumer::Consumer(consumer_id, _) => {
+ tracing::trace!(
+ "Getting next messages for consumer id: {} for partition: {} from offset: {}...",
+ consumer_id,
+ partition_id,
+ offset
+ );
+ }
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
+ tracing::trace!(
+ "Getting next messages for consumer group: {} member: {} for partition: {} from offset: {}...",
+ consumer_group_id.0,
+ member_id.0,
+ partition_id,
+ offset
+ );
+ }
+ }
let batches = self
.get_messages_by_offset(stream_id, topic_id, partition_id, offset, count)
.await?;
@@ -1406,6 +1419,14 @@ impl Streams {
match consumer {
PollingConsumer::Consumer(consumer_id, _) => {
+ tracing::trace!(
+ "Auto-committing offset {} for consumer {} on stream {}, topic {}, partition {}",
+ offset,
+ consumer_id,
+ numeric_stream_id,
+ numeric_topic_id,
+ partition_id
+ );
let (offset_value, path) = self.with_partition_by_id(
stream_id,
topic_id,
@@ -1427,18 +1448,33 @@ impl Streams {
);
crate::streaming::partitions::storage::persist_offset(&path, offset_value).await?;
}
- PollingConsumer::ConsumerGroup(cg_id, _) => {
+ PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
+ tracing::trace!(
+ "Auto-committing offset {} for consumer group {} member {} on stream {}, topic {}, partition {}",
+ offset,
+ consumer_group_id.0,
+ member_id.0,
+ numeric_stream_id,
+ numeric_topic_id,
+ partition_id
+ );
let (offset_value, path) = self.with_partition_by_id(
stream_id,
topic_id,
partition_id,
|(.., offsets, _)| {
let hdl = offsets.pin();
+ let key = (consumer_group_id, member_id);
let item = hdl.get_or_insert(
- cg_id,
- crate::streaming::partitions::consumer_offset::ConsumerOffset::default_for_consumer_group(
- cg_id as u32,
- &config.get_consumer_group_offsets_path(numeric_stream_id, numeric_topic_id, partition_id),
+ key,
+ ConsumerOffset::default_for_consumer_group(
+ consumer_group_id,
+ member_id,
+ &config.get_consumer_group_offsets_path(
+ numeric_stream_id,
+ numeric_topic_id,
+ partition_id,
+ ),
),
);
item.offset.store(offset, Ordering::Relaxed);
diff --git a/core/server/src/streaming/partitions/consumer_offset.rs b/core/server/src/streaming/partitions/consumer_offset.rs
index 4f88e9906..bd10014f8 100644
--- a/core/server/src/streaming/partitions/consumer_offset.rs
+++ b/core/server/src/streaming/partitions/consumer_offset.rs
@@ -17,6 +17,7 @@
use std::sync::atomic::AtomicU64;
+use crate::streaming::polling_consumer::{ConsumerGroupId, MemberId};
use iggy_common::ConsumerKind;
#[derive(Debug)]
@@ -48,12 +49,16 @@ impl ConsumerOffset {
}
}
- pub fn default_for_consumer_group(consumer_id: u32, path: &str) -> Self {
+ pub fn default_for_consumer_group(
+ consumer_group_id: ConsumerGroupId,
+ member_id: MemberId,
+ path: &str,
+ ) -> Self {
Self {
kind: ConsumerKind::ConsumerGroup,
- consumer_id,
+ consumer_id: member_id.0 as u32,
offset: AtomicU64::new(0),
- path: format!("{path}/{consumer_id}"),
+ path: format!("{path}/{}_{}", consumer_group_id.0, member_id.0),
}
}
diff --git a/core/server/src/streaming/partitions/helpers.rs b/core/server/src/streaming/partitions/helpers.rs
index 549156eee..e8e7487b6 100644
--- a/core/server/src/streaming/partitions/helpers.rs
+++ b/core/server/src/streaming/partitions/helpers.rs
@@ -38,6 +38,7 @@ use crate::{
partition::{self, PartitionRef, PartitionRefMut},
storage,
},
+ polling_consumer::{ConsumerGroupId, MemberId},
segments::{IggyIndexesMut, IggyMessagesBatchMut, IggyMessagesBatchSet, storage::Storage},
},
};
@@ -137,10 +138,12 @@ pub fn get_consumer_offset(
}
pub fn get_consumer_group_member_offset(
- id: usize,
+ consumer_group_id: ConsumerGroupId,
+ member_id: MemberId,
) -> impl FnOnce(ComponentsById<PartitionRef>) -> Option<ConsumerOffsetInfo> {
move |(root, _, _, current_offset, _, offsets, _)| {
- offsets.pin().get(&id).map(|item| ConsumerOffsetInfo {
+ let key = (consumer_group_id, member_id);
+ offsets.pin().get(&key).map(|item| ConsumerOffsetInfo {
partition_id: root.id() as u32,
current_offset: current_offset.load(Ordering::Relaxed),
stored_offset: item.offset.load(Ordering::Relaxed),
@@ -208,7 +211,8 @@ pub fn delete_consumer_offset_from_disk(
}
pub fn store_consumer_group_member_offset(
- id: usize,
+ consumer_group_id: ConsumerGroupId,
+ member_id: MemberId,
stream_id: usize,
topic_id: usize,
partition_id: usize,
@@ -217,10 +221,12 @@ pub fn store_consumer_group_member_offset(
) -> impl FnOnce(ComponentsById<PartitionRef>) {
move |(.., offsets, _)| {
let hdl = offsets.pin();
+ let key = (consumer_group_id, member_id);
let item = hdl.get_or_insert(
- id,
+ key,
ConsumerOffset::default_for_consumer_group(
- id as u32,
+ consumer_group_id,
+ member_id,
&config.get_consumer_group_offsets_path(stream_id, topic_id, partition_id),
),
);
@@ -229,24 +235,28 @@ pub fn store_consumer_group_member_offset(
}
pub fn delete_consumer_group_member_offset(
- id: usize,
+ consumer_group_id: ConsumerGroupId,
+ member_id: MemberId,
) -> impl FnOnce(ComponentsById<PartitionRef>) -> Result<String, IggyError> {
move |(.., offsets, _)| {
let hdl = offsets.pin();
+ let key = (consumer_group_id, member_id);
let offset = hdl
- .remove(&id)
- .ok_or_else(|| IggyError::ConsumerOffsetNotFound(id))?;
+ .remove(&key)
+ .ok_or_else(|| IggyError::ConsumerOffsetNotFound(member_id.0))?;
Ok(offset.path.clone())
}
}
pub fn persist_consumer_group_member_offset_to_disk(
- id: usize,
+ consumer_group_id: ConsumerGroupId,
+ member_id: MemberId,
) -> impl AsyncFnOnce(ComponentsById<PartitionRef>) -> Result<(), IggyError> {
async move |(.., offsets, _)| {
let hdl = offsets.pin();
+ let key = (consumer_group_id, member_id);
let item = hdl
- .get(&id)
+ .get(&key)
.expect("persist_consumer_group_member_offset_to_disk: offset not found");
let offset = item.offset.load(Ordering::Relaxed);
storage::persist_offset(&item.path, offset).await
@@ -254,12 +264,14 @@ pub fn persist_consumer_group_member_offset_to_disk(
}
pub fn delete_consumer_group_member_offset_from_disk(
- id: usize,
+ consumer_group_id: ConsumerGroupId,
+ member_id: MemberId,
) -> impl AsyncFnOnce(ComponentsById<PartitionRef>) -> Result<(), IggyError> {
async move |(.., offsets, _)| {
let hdl = offsets.pin();
+ let key = (consumer_group_id, member_id);
let item = hdl
- .get(&id)
+ .get(&key)
.expect("delete_consumer_group_member_offset_from_disk: offset not found");
let path = &item.path;
storage::delete_persisted_offset(path).await
diff --git a/core/server/src/streaming/partitions/partition.rs b/core/server/src/streaming/partitions/partition.rs
index 4dc42f0d2..5875b9c69 100644
--- a/core/server/src/streaming/partitions/partition.rs
+++ b/core/server/src/streaming/partitions/partition.rs
@@ -29,6 +29,7 @@ use crate::{
consumer_offset, helpers::create_message_deduplicator, journal::MemoryMessageJournal,
log::SegmentedLog,
},
+ polling_consumer::{ConsumerGroupId, MemberId},
stats::{PartitionStats, TopicStats},
},
};
@@ -69,7 +70,9 @@ impl std::ops::DerefMut for ConsumerOffsets {
}
#[derive(Debug, Clone)]
-pub struct ConsumerGroupOffsets(papaya::HashMap<usize, consumer_offset::ConsumerOffset>);
+pub struct ConsumerGroupOffsets(
+ papaya::HashMap<(ConsumerGroupId, MemberId), consumer_offset::ConsumerOffset>,
+);
impl ConsumerGroupOffsets {
pub fn with_capacity(capacity: usize) -> Self {
@@ -79,7 +82,7 @@ impl ConsumerGroupOffsets {
impl<I> From<I> for ConsumerGroupOffsets
where
- I: IntoIterator<Item = (usize, consumer_offset::ConsumerOffset)>,
+ I: IntoIterator<Item = ((ConsumerGroupId, MemberId), consumer_offset::ConsumerOffset)>,
{
fn from(iter: I) -> Self {
Self(papaya::HashMap::from_iter(iter))
@@ -87,7 +90,7 @@ where
}
impl std::ops::Deref for ConsumerGroupOffsets {
- type Target = papaya::HashMap<usize, consumer_offset::ConsumerOffset>;
+ type Target = papaya::HashMap<(ConsumerGroupId, MemberId), consumer_offset::ConsumerOffset>;
fn deref(&self) -> &Self::Target {
&self.0
diff --git a/core/server/src/streaming/partitions/storage.rs b/core/server/src/streaming/partitions/storage.rs
index 9857fa860..bafa382e2 100644
--- a/core/server/src/streaming/partitions/storage.rs
+++ b/core/server/src/streaming/partitions/storage.rs
@@ -17,8 +17,10 @@
use super::COMPONENT;
use crate::{
- configs::system::SystemConfig, io::fs_utils::remove_dir_all,
+ configs::system::SystemConfig,
+ io::fs_utils::remove_dir_all,
streaming::partitions::consumer_offset::ConsumerOffset,
+ streaming::polling_consumer::{ConsumerGroupId, MemberId},
};
use compio::{
fs::{self, OpenOptions, create_dir_all},
@@ -159,10 +161,7 @@ pub async fn persist_offset(path: &str, offset: u64) -> Result<(), IggyError> {
Ok(())
}
-pub fn load_consumer_offsets(
- path: &str,
- kind: ConsumerKind,
-) -> Result<Vec<ConsumerOffset>, IggyError> {
+pub fn load_consumer_offsets(path: &str) -> Result<Vec<ConsumerOffset>, IggyError> {
trace!("Loading consumer offsets from path: {path}...");
let dir_entries = std::fs::read_dir(path);
if dir_entries.is_err() {
@@ -183,11 +182,9 @@ pub fn load_consumer_offsets(
}
let name = dir_entry.file_name().into_string().unwrap();
- let consumer_id = name.parse::<u32>();
- if consumer_id.is_err() {
- error!("Invalid consumer ID file with name: '{}'.", name);
- continue;
- }
+ let consumer_id = name.parse::<u32>().unwrap_or_else(|_| {
+ panic!("Invalid consumer ID file with name: '{}'.", name);
+ });
let path = dir_entry.path();
let path = path.to_str();
@@ -197,7 +194,6 @@ pub fn load_consumer_offsets(
}
let path = path.unwrap().to_string();
- let consumer_id = consumer_id.unwrap();
let file = std::fs::File::open(&path)
.with_error(|error| {
format!("{COMPONENT} (error: {error}) - failed to open offset file, path: {path}")
@@ -214,7 +210,7 @@ pub fn load_consumer_offsets(
let offset = AtomicU64::new(u64::from_le_bytes(offset));
consumer_offsets.push(ConsumerOffset {
- kind,
+ kind: ConsumerKind::Consumer,
consumer_id,
offset,
path,
@@ -224,3 +220,93 @@ pub fn load_consumer_offsets(
consumer_offsets.sort_by(|a, b| a.consumer_id.cmp(&b.consumer_id));
Ok(consumer_offsets)
}
+
+pub fn load_consumer_group_offsets(
+ path: &str,
+) -> Result<Vec<(ConsumerGroupId, MemberId, ConsumerOffset)>, IggyError> {
+ trace!("Loading consumer group offsets from path: {path}...");
+ let dir_entries = std::fs::read_dir(path);
+ if dir_entries.is_err() {
+ return Err(IggyError::CannotReadConsumerOffsets(path.to_owned()));
+ }
+
+ let mut consumer_group_offsets = Vec::new();
+ let dir_entries = dir_entries.unwrap();
+ for dir_entry in dir_entries {
+ let dir_entry = dir_entry.unwrap();
+ let metadata = dir_entry.metadata();
+ if metadata.is_err() {
+ break;
+ }
+
+ if metadata.unwrap().is_dir() {
+ continue;
+ }
+
+ let name = dir_entry.file_name().into_string().unwrap();
+
+ // For consumer groups, filename format is: {consumer_group_id}_{member_id}
+ let (consumer_group_id, member_id) = if let Some(underscore_pos) = name.find('_') {
+ let consumer_group_id_str = &name[..underscore_pos];
+ let member_id_str = &name[underscore_pos + 1..];
+ let consumer_group_id = consumer_group_id_str.parse::<u32>().unwrap_or_else(|_| {
+ panic!(
+ "Invalid consumer group ID in consumer group file with name: '{}'.",
+ name
+ );
+ });
+ let member_id = member_id_str.parse::<u32>().unwrap_or_else(|_| {
+ panic!(
+ "Invalid member ID in consumer group file with name: '{}'.",
+ name
+ );
+ });
+ (
+ ConsumerGroupId(consumer_group_id as usize),
+ MemberId(member_id as usize),
+ )
+ } else {
+ panic!(
+ "Invalid consumer group file format with name: '{}'. Expected format: 'group_id_member_id'.",
+ name
+ );
+ };
+
+ let path = dir_entry.path();
+ let path = path.to_str();
+ if path.is_none() {
+ error!(
+ "Invalid consumer group offset path for file with name: '{}'.",
+ name
+ );
+ continue;
+ }
+
+ let path = path.unwrap().to_string();
+ let file = std::fs::File::open(&path)
+ .with_error(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to open offset file, path: {path}")
+ })
+ .map_err(|_| IggyError::CannotReadFile)?;
+ let mut cursor = std::io::Cursor::new(file);
+ let mut offset = [0; 8];
+ cursor
+ .get_mut().read_exact(&mut offset)
+ .with_error(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to read consumer group offset from file, path: {path}")
+ })
+ .map_err(|_| IggyError::CannotReadFile)?;
+ let offset = AtomicU64::new(u64::from_le_bytes(offset));
+
+ let consumer_offset = ConsumerOffset {
+ kind: ConsumerKind::ConsumerGroup,
+ consumer_id: member_id.0 as u32,
+ offset,
+ path,
+ };
+
+ consumer_group_offsets.push((consumer_group_id, member_id, consumer_offset));
+ }
+
+ Ok(consumer_group_offsets)
+}
diff --git a/core/server/src/streaming/polling_consumer.rs b/core/server/src/streaming/polling_consumer.rs
index 2f75f379e..dcc5f0f4a 100644
--- a/core/server/src/streaming/polling_consumer.rs
+++ b/core/server/src/streaming/polling_consumer.rs
@@ -20,10 +20,28 @@ use crate::streaming::utils::hash;
use iggy_common::{IdKind, Identifier};
use std::fmt::{Display, Formatter};
+#[derive(Debug, PartialEq, Eq, Hash, Copy, Clone)]
+pub struct ConsumerGroupId(pub usize);
+
+impl Display for ConsumerGroupId {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}", self.0)
+ }
+}
+
+#[derive(Debug, PartialEq, Eq, Hash, Copy, Clone)]
+pub struct MemberId(pub usize);
+
+impl Display for MemberId {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}", self.0)
+ }
+}
+
#[derive(Debug, PartialEq, Copy, Clone)]
pub enum PollingConsumer {
- Consumer(usize, usize), // Consumer ID + Partition ID
- ConsumerGroup(usize, usize), // Consumer Group ID + Member ID
+ Consumer(usize, usize), // Consumer ID + Partition ID
+ ConsumerGroup(ConsumerGroupId, MemberId), // Consumer Group ID + Member ID
}
impl PollingConsumer {
@@ -32,7 +50,7 @@ impl PollingConsumer {
}
pub fn consumer_group(consumer_group_id: usize, member_id: usize) -> Self {
- PollingConsumer::ConsumerGroup(consumer_group_id, member_id)
+ PollingConsumer::ConsumerGroup(ConsumerGroupId(consumer_group_id), MemberId(member_id))
}
pub fn resolve_consumer_id(identifier: &Identifier) -> usize {
@@ -103,8 +121,8 @@ mod tests {
match polling_consumer {
PollingConsumer::ConsumerGroup(consumer_group_id, member_id) => {
- assert_eq!(consumer_group_id, group_id);
- assert_eq!(member_id, client_id);
+ assert_eq!(consumer_group_id, ConsumerGroupId(group_id));
+ assert_eq!(member_id, MemberId(client_id));
}
_ => panic!("Expected ConsumerGroup"),
}
diff --git a/core/server/src/streaming/streams/helpers.rs b/core/server/src/streaming/streams/helpers.rs
index ad146bb63..98eaf2d14 100644
--- a/core/server/src/streaming/streams/helpers.rs
+++ b/core/server/src/streaming/streams/helpers.rs
@@ -81,7 +81,8 @@ pub fn store_consumer_offset(
}
pub fn store_consumer_group_member_offset(
- member_id: usize,
+ consumer_group_id: crate::streaming::polling_consumer::ConsumerGroupId,
+ member_id: crate::streaming::polling_consumer::MemberId,
topic_id: &Identifier,
partition_id: usize,
offset: u64,
@@ -94,6 +95,7 @@ pub fn store_consumer_group_member_offset(
root.partitions().with_components_by_id(
partition_id,
partitions::helpers::store_consumer_group_member_offset(
+ consumer_group_id,
member_id,
stream_id,
topic_id,