Improve partition ID calculation (#1236)

iggy-rs · Sep 11, 2024 · 4149a19 · 4149a19
1 parent 638dab6
commit 4149a19
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 5 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/server/Cargo.toml b/server/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.4.40"
+version = "0.4.41"
 edition = "2021"
 build = "src/build.rs"
 

diff --git a/server/src/streaming/topics/consumer_group.rs b/server/src/streaming/topics/consumer_group.rs
@@ -131,9 +131,18 @@ impl ConsumerGroupMember {
 
  pub fn calculate_partition_id(&mut self) -> u32 {
  let partition_index = self.current_partition_index;
- let partition_id = *self.partitions.get(&partition_index).unwrap();
+ let partition_id = if let Some(partition_id) = self.partitions.get(&partition_index) {
+ *partition_id
+ } else {
+ trace!(
+ "No partition ID found for index: {} for member with ID: {}.",
+ partition_index,
+ self.id
+ );
+ return 1;
+ };
  self.current_partition_id = partition_id;
- if self.partitions.len() == (partition_index + 1) as usize {
+ if self.partitions.len() <= (partition_index + 1) as usize {
  self.current_partition_index = 0;
  } else {
  self.current_partition_index += 1;

diff --git a/server/src/tcp/tcp_listener.rs b/server/src/tcp/tcp_listener.rs
@@ -14,7 +14,7 @@ pub async fn start(address: &str, system: SharedSystem) -> SocketAddr {
  tokio::spawn(async move {
  let listener = TcpListener::bind(&address)
  .await
- .expect("Unable to start TCP TLS server.");
+ .expect("Unable to start TCP server.");
 
  let local_addr = listener
  .local_addr()