fix(server): fix & remove unnecessary error logs on the server, updat…

…e deps
apache · Feb 20, 2025 · a1b2c7a · a1b2c7a
1 parent 7c7f2e5
commit a1b2c7a
Show file tree

Hide file tree

Showing 43 changed files with 335 additions and 279 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bench/Cargo.toml b/bench/Cargo.toml
@@ -21,7 +21,7 @@ iggy = { path = "../sdk" }
 iggy-bench-report = { path = "report" }
 integration = { path = "../integration" }
 nonzero_lit = "0.1.2"
-serde = { version = "1.0.217", features = ["derive"] }
+serde = { version = "1.0.218", features = ["derive"] }
 sysinfo = "0.33.1"
 tokio = { version = "1.43.0", features = ["full"] }
 toml = "0.8.20"

diff --git a/cli/Cargo.toml b/cli/Cargo.toml
@@ -16,7 +16,7 @@ login-session = ["dep:keyring"]
 
 [dependencies]
 ahash = { version = "0.8.11", features = ["serde"] }
-anyhow = "1.0.95"
+anyhow = "1.0.96"
 clap = { version = "4.5.30", features = ["derive"] }
 clap_complete = "4.5.45"
 figlet-rs = "0.1.5"

diff --git a/examples/Cargo.toml b/examples/Cargo.toml
@@ -74,14 +74,14 @@ path = "src/stream-builder/stream-producer-config/main.rs"
 
 [dependencies]
 ahash = { version = "0.8.11", features = ["serde"] }
-anyhow = "1.0.95"
+anyhow = "1.0.96"
 bytes = "1.10.0"
 clap = { version = "4.5.30", features = ["derive"] }
 futures-util = "0.3.31"
 iggy = { path = "../sdk" }
 rand = "0.9.0"
-serde = { version = "1.0.217", features = ["derive", "rc"] }
-serde_json = "1.0.138"
+serde = { version = "1.0.218", features = ["derive", "rc"] }
+serde_json = "1.0.139"
 tokio = { version = "1.43.0", features = ["full"] }
 tracing = { version = "0.1.41" }
 tracing-subscriber = { version = "0.3.19", features = ["fmt", "env-filter"] }
diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml
@@ -14,7 +14,7 @@ readme = "../README.md"
 [dependencies]
 aes-gcm = "0.10.3"
 ahash = { version = "0.8.11", features = ["serde"] }
-anyhow = "1.0.95"
+anyhow = "1.0.96"
 async-broadcast = { version = "0.7.2" }
 async-dropper = { version = "0.3.1", features = ["tokio", "simple"] }
 async-trait = "0.1.86"
@@ -51,8 +51,8 @@ reqwest = { version = "0.12.12", default-features = false, features = [
 reqwest-middleware = { version = "0.4.0", features = ["json"] }
 reqwest-retry = "0.7.0"
 rustls = { version = "0.23.23", features = ["ring"] }
-serde = { version = "1.0.217", features = ["derive", "rc"] }
-serde_json = "1.0.138"
+serde = { version = "1.0.218", features = ["derive", "rc"] }
+serde_json = "1.0.139"
 serde_with = { version = "3.12.0", features = ["base64"] }
 strum = { version = "0.27.1", features = ["derive"] }
 thiserror = "2.0.11"
@@ -66,8 +66,8 @@ webpki-roots = { version = "0.26.8" }
 
 [build-dependencies]
 convert_case = "0.7.1"
-serde = { version = "1.0.217", features = ["derive", "rc"] }
-serde_derive = "1.0.217"
+serde = { version = "1.0.218", features = ["derive", "rc"] }
+serde_derive = "1.0.218"
 
 [features]
 default = ["tokio_lock"]

diff --git a/server/Cargo.toml b/server/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.4.209"
+version = "0.4.210"
 edition = "2021"
 build = "src/build.rs"
 license = "Apache-2.0"
@@ -13,7 +13,7 @@ mimalloc = ["dep:mimalloc"]
 
 [dependencies]
 ahash = { version = "0.8.11" }
-anyhow = "1.0.95"
+anyhow = "1.0.96"
 async_zip = { version = "0.0.17", features = [
     "tokio",
     "lzma",
@@ -76,8 +76,8 @@ ring = "0.17.9"
 rust-s3 = { version = "0.35.1", features = ["default"] }
 rustls = { version = "0.23.23" }
 rustls-pemfile = "2.2.0"
-serde = { version = "1.0.217", features = ["derive", "rc"] }
-serde_json = "1.0.138"
+serde = { version = "1.0.218", features = ["derive", "rc"] }
+serde_json = "1.0.139"
 serde_with = { version = "3.12.0", features = ["base64", "macros"] }
 static-toml = "1.3.0"
 strum = { version = "0.27.1", features = ["derive"] }
@@ -106,7 +106,7 @@ mockall = "0.13.1"
 
 [build-dependencies]
 figment = { version = "0.10.19", features = ["json", "toml", "env"] }
-serde_json = "1.0.138"
+serde_json = "1.0.139"
 vergen-git2 = { version = "1.0.5", features = [
     "build",
     "cargo",

diff --git a/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs b/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
@@ -26,8 +26,8 @@ pub async fn handle(
                 &command.group_id,
             )
             .await.with_error_context(|_| format!(
-                "{COMPONENT} - failed to delete consumer group for stream_id: {}, topic_id: {}, group_id: {:?}, session: {}",
-                command.stream_id, command.topic_id, command.group_id, session
+                "{COMPONENT} - failed to delete consumer group with ID: {} for topic with ID: {} in stream with ID: {} for session: {}",
+                command.group_id, command.topic_id, command.stream_id, session
             ))?;
     }
 

diff --git a/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs b/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
@@ -15,18 +15,18 @@ pub async fn handle(
 ) -> Result<(), IggyError> {
     debug!("session: {session}, command: {command}");
     let system = system.read().await;
-    let consumer_group = system.get_consumer_group(
+    let Some(consumer_group) = system.get_consumer_group(
         session,
         &command.stream_id,
         &command.topic_id,
         &command.group_id,
-    );
-    if consumer_group.is_err() {
+    )?
+    else {
         sender.send_empty_ok_response().await?;
         return Ok(());
-    }
+    };
 
-    let consumer_group = consumer_group?.read().await;
+    let consumer_group = consumer_group.read().await;
     let consumer_group = mapper::map_consumer_group(&consumer_group).await;
     sender.send_ok_response(&consumer_group).await?;
     Ok(())

diff --git a/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs b/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
@@ -25,8 +25,8 @@ pub async fn handle(
             command.partition_id,
         )
         .await
-        .with_error_context(|_| format!("{COMPONENT} - failed to delete consumer offset for stream_id: {}, topic_id: {}, partition_id: {:?}, session: {}",
-            command.stream_id, command.topic_id, command.partition_id, session
+        .with_error_context(|_| format!("{COMPONENT} - failed to delete consumer offset for topic with ID: {} in stream with ID: {} partition ID: {:#?}, session: {}",
+            command.topic_id, command.stream_id, command.partition_id, session
         ))?;
     sender.send_empty_ok_response().await?;
     Ok(())

diff --git a/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs b/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
@@ -15,21 +15,16 @@ pub async fn handle(
 ) -> Result<(), IggyError> {
     debug!("session: {session}, command: {command}");
     let system = system.read().await;
-    let offset = system
+    let Some(offset) = system
         .get_consumer_offset(
             session,
             &command.consumer,
             &command.stream_id,
             &command.topic_id,
             command.partition_id,
         )
-        .await;
-    if offset.is_err() {
-        sender.send_empty_ok_response().await?;
-        return Ok(());
-    }
-
-    let Some(offset) = offset? else {
+        .await?
+    else {
         sender.send_empty_ok_response().await?;
         return Ok(());
     };

diff --git a/server/src/binary/handlers/partitions/delete_partitions_handler.rs b/server/src/binary/handlers/partitions/delete_partitions_handler.rs
@@ -31,8 +31,7 @@ pub async fn handle(
             .await
             .with_error_context(|_| {
                 format!(
-                    "{COMPONENT} - failed to delete partitions for stream_id: {}, topic_id: {}, session: {}",
-                    stream_id, topic_id, session
+                    "{COMPONENT} - failed to delete partitions for topic with ID: {topic_id} in stream with ID: {stream_id}, session: {session}",
                 )
             })?;
     }

diff --git a/server/src/binary/handlers/streams/delete_stream_handler.rs b/server/src/binary/handlers/streams/delete_stream_handler.rs
@@ -23,7 +23,7 @@ pub async fn handle(
             .delete_stream(session, &command.stream_id)
             .await
             .with_error_context(|_| {
-                format!("{COMPONENT} - failed to delete stream with id: {stream_id}, session: {session}")
+                format!("{COMPONENT} - failed to delete stream with ID: {stream_id}, session: {session}")
             })?;
     }
 
@@ -33,7 +33,7 @@ pub async fn handle(
         .apply(session.get_user_id(), EntryCommand::DeleteStream(command))
         .await
         .with_error_context(|_| {
-            format!("{COMPONENT} - failed to apply delete stream with id: {stream_id}, session: {session}")
+            format!("{COMPONENT} - failed to apply delete stream with ID: {stream_id}, session: {session}")
         })?;
     sender.send_empty_ok_response().await?;
     Ok(())