Merge pull request #274 from influxdata/crepererum/update-rust-and-fix-depr

chore: update rust, fix deprecations & lints

Author: crepererum

Cargo.toml

@@ -2,7 +2,7 @@
 name = "rskafka"
 version = "0.6.0"
 edition = "2024"
-rust-version = "1.85"
+rust-version = "1.88"
 license = "MIT OR Apache-2.0"
 readme = "README.md"
 keywords = [

fuzz/fuzz_targets/protocol_reader.rs

@@ -104,7 +104,7 @@ fn driver(data: &[u8]) -> Result<(), Error> {
             api_key,
             api_version,
         ),
-        _ => Err(format!("Fuzzing not implemented for: {:?}", api_key).into()),
+        _ => Err(format!("Fuzzing not implemented for: {api_key:?}").into()),
     }
 }
 

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.85"
+channel = "1.88"
 components = [ "rustfmt", "clippy" ]

src/backoff.rs

@@ -175,7 +175,6 @@ impl Iterator for Backoff {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use rand::rngs::mock::StepRng;
 
     #[test]
     fn test_backoff() {
@@ -190,18 +189,18 @@ mod tests {
             deadline: None,
         };
 
-        let assert_fuzzy_eq = |a: f64, b: f64| assert!((b - a).abs() < 0.0001, "{} != {}", a, b);
+        let assert_fuzzy_eq = |a: f64, b: f64| assert!((b - a).abs() < 0.0001, "{a} != {b}");
 
         // Create a static rng that takes the minimum of the range
-        let rng = Box::new(StepRng::new(0, 0));
+        let rng = Box::new(ConstantRng::new(0));
         let mut backoff = Backoff::new_with_rng(&config, Some(rng));
 
         for _ in 0..20 {
             assert_eq!(backoff.next().unwrap().as_secs_f64(), init_backoff_secs);
         }
 
         // Create a static rng that takes the maximum of the range
-        let rng = Box::new(StepRng::new(u64::MAX, 0));
+        let rng = Box::new(ConstantRng::new(u64::MAX));
         let mut backoff = Backoff::new_with_rng(&config, Some(rng));
 
         for i in 0..20 {
@@ -210,7 +209,7 @@ mod tests {
         }
 
         // Create a static rng that takes the mid point of the range
-        let rng = Box::new(StepRng::new(u64::MAX / 2, 0));
+        let rng = Box::new(ConstantRng::new(u64::MAX / 2));
         let mut backoff = Backoff::new_with_rng(&config, Some(rng));
 
         let mut value = init_backoff_secs;
@@ -221,7 +220,7 @@ mod tests {
         }
 
         // deadline
-        let rng = Box::new(StepRng::new(u64::MAX, 0));
+        let rng = Box::new(ConstantRng::new(u64::MAX));
         let deadline = Duration::from_secs_f64(init_backoff_secs);
         let mut backoff = Backoff::new_with_rng(
             &BackoffConfig {
@@ -232,4 +231,29 @@ mod tests {
         );
         assert_eq!(backoff.next(), None);
     }
+
+    /// A simple RNG that always returns the same value
+    struct ConstantRng {
+        value: u64,
+    }
+
+    impl ConstantRng {
+        fn new(value: u64) -> Self {
+            Self { value }
+        }
+    }
+
+    impl RngCore for ConstantRng {
+        fn next_u32(&mut self) -> u32 {
+            (self.value >> 32) as u32
+        }
+
+        fn next_u64(&mut self) -> u64 {
+            self.value
+        }
+
+        fn fill_bytes(&mut self, _dest: &mut [u8]) {
+            unimplemented!()
+        }
+    }
 }

src/client/consumer.rs

@@ -448,7 +448,7 @@ mod tests {
                     buffered += size;
                 }
 
-                println!("Waiting up to {} ms for more data", max_wait_ms);
+                println!("Waiting up to {max_wait_ms} ms for more data");
 
                 // Need to wait for more data
                 let timeout = tokio::time::sleep(Duration::from_millis(max_wait_ms as u64)).fuse();

src/client/controller.rs

@@ -186,8 +186,7 @@ impl BrokerCache for &ControllerClient {
         let controller_id = self.get_controller_id().await?;
         let broker = self.brokers.connect(controller_id).await?.ok_or_else(|| {
             Error::InvalidResponse(format!(
-                "Controller {} not found in metadata response",
-                controller_id
+                "Controller {controller_id} not found in metadata response"
             ))
         })?;
 

src/client/partition.rs

@@ -450,8 +450,7 @@ impl BrokerCache for &PartitionClient {
                     );
                 }
                 Err(Error::InvalidResponse(format!(
-                    "Partition leader {} not found in metadata response",
-                    leader
+                    "Partition leader {leader} not found in metadata response"
                 )))
             }
             Err(e) => {
@@ -972,8 +971,7 @@ fn extract_offset(partition: ListOffsetsResponsePartition) -> Result<i64> {
             Some(offsets) => match offsets.len() {
                 1 => Ok(offsets[0].0),
                 n => Err(Error::InvalidResponse(format!(
-                    "Expected 1 offset to be returned but got {}",
-                    n
+                    "Expected 1 offset to be returned but got {n}"
                 ))),
             },
             None => Err(Error::InvalidResponse(

src/messenger.rs

@@ -633,7 +633,7 @@ fn sorted_ranges_repr(ranges: &HashMap<ApiKey, ApiVersionRange>) -> String {
     ranges.sort_by_key(|(key, _range)| *key);
     let ranges: Vec<_> = ranges
         .into_iter()
-        .map(|(key, range)| format!("{:?}: {}", key, range))
+        .map(|(key, range)| format!("{key:?}: {range}"))
         .collect();
     ranges.join(", ")
 }

src/protocol/error.rs

@@ -357,7 +357,7 @@ impl From<Option<Error>> for Int16 {
 
 impl std::fmt::Display for Error {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{:?}", self)
+        write!(f, "{self:?}")
     }
 }
 

src/protocol/messages/mod.rs

@@ -140,7 +140,7 @@ fn read_versioned_array<R: Read, T: ReadVersionedType<R>>(
     match len {
         -1 => Ok(None),
         l if l < -1 => Err(ReadVersionedError::ReadError(ReadError::Malformed(
-            format!("Invalid negative length for array: {}", l).into(),
+            format!("Invalid negative length for array: {l}").into(),
         ))),
         _ => {
             let len = usize::try_from(len).map_err(ReadError::Overflow)?;