kjuulh/nodata
1
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: remove more partitions

Browse Source 
Signed-off-by: kjuulh <contact@kjuulh.io>
This commit is contained in:
Kasper Juul Hermansen 2024-11-09 16:05:26 +01:00
parent 6dfd7d7011
commit ecc6d785e7
No known key found for this signature in database
1 changed files with 4 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
crates/nodata/src/services/consumers.rs
View File

@ -28,7 +28,6 @@ pub struct ConsumerGroup {
pub struct InnerConsumerGroup { pub struct InnerConsumerGroup {
pub id: ConsumerId, pub id: ConsumerId,
topic: Topic, topic: Topic,
partition_key: Option<PartitionKey>,
pub(crate) offset: TopicOffset, pub(crate) offset: TopicOffset,
pub(crate) cur_offset: TopicOffset, pub(crate) cur_offset: TopicOffset,
@ -36,17 +35,9 @@ pub struct InnerConsumerGroup {
} }
impl ConsumerGroup { impl ConsumerGroup {
pub fn new( pub fn new(id: impl Into<ConsumerId>, topic: impl Into<Topic>) -> Self {
id: impl Into<ConsumerId>,
topic: impl Into<Topic>,
partition_key: Option<impl Into<PartitionKey>>,
) -> Self {
Self { Self {
inner: Arc::new(RwLock::new(InnerConsumerGroup::new( inner: Arc::new(RwLock::new(InnerConsumerGroup::new(id, topic))),
id,
topic,
partition_key,
))),
} }
} }
@ -83,15 +74,10 @@ impl ConsumerGroup {
} }
impl InnerConsumerGroup { impl InnerConsumerGroup {
pub fn new( pub fn new(id: impl Into<ConsumerId>, topic: impl Into<Topic>) -> Self {
id: impl Into<ConsumerId>,
topic: impl Into<Topic>,
partition_key: Option<impl Into<PartitionKey>>,
) -> Self {
Self { Self {
id: id.into(), id: id.into(),
topic: topic.into(), topic: topic.into(),
partition_key: partition_key.map(|pk| pk.into()),
offset: 0, offset: 0,
cur_offset: 0, cur_offset: 0,
members: BTreeMap::default(), members: BTreeMap::default(),
@ -181,10 +167,7 @@ impl Consumers {
let mut storage = self.storage.write().await; let mut storage = self.storage.write().await;
if !storage.contains_key(&id) { if !storage.contains_key(&id) {
storage.insert( storage.insert(id.clone(), ConsumerGroup::new(&id, &topic));
id.clone(),
ConsumerGroup::new(&id, &topic, None::<PartitionKey>),
);
} }
let consumer_group = storage.get_mut(&id).unwrap(); let consumer_group = storage.get_mut(&id).unwrap();