Merge branch 'main' into split_produce_requests

shotover · Sep 10, 2024 · 5c4d7fb · 5c4d7fb
2 parents 4ab407c + 7ce587b
commit 5c4d7fb
Show file tree

Hide file tree

Showing 16 changed files with 41 additions and 837 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
@@ -1,4 +1,4 @@
 [toolchain]
-channel = "1.80"
+channel = "1.81"
 components = [ "rustfmt", "clippy" ]
 targets = [ "aarch64-unknown-linux-gnu" ]
diff --git a/shotover-proxy/benches/windsock/cassandra/bench.rs b/shotover-proxy/benches/windsock/cassandra/bench.rs
@@ -74,9 +74,9 @@ pub enum CassandraDb {
 }
 
 enum CassandraDbInstance {
-    #[allow(dead_code)] // must be held to delay drop
+    #[expect(dead_code, reason = "must be held to delay drop")]
     Compose(DockerCompose),
-    #[allow(dead_code)]
+    #[expect(dead_code)]
     Mocked(MockHandle),
 }
 
@@ -368,7 +368,7 @@ pub struct CassandraBench {
 }
 
 impl CassandraBench {
-    #[allow(clippy::too_many_arguments)]
+    #[expect(clippy::too_many_arguments)]
     pub fn new(
         db: CassandraDb,
         topology: CassandraTopology,

diff --git a/shotover-proxy/tests/redis_int_tests/basic_driver_tests.rs b/shotover-proxy/tests/redis_int_tests/basic_driver_tests.rs
@@ -1093,9 +1093,7 @@ fn assert_cluster_ports_rewrite_nodes(res: Value, new_port: u16) {
     for result in reader.records() {
         let record = result.unwrap();
 
-        let port = record[1]
-            .split(|c| c == ':' || c == '@')
-            .collect::<Vec<&str>>();
+        let port = record[1].split([':', '@']).collect::<Vec<&str>>();
 
         assert_eq!(port[1].parse::<u16>().unwrap(), new_port);
         assertion_run = true;

diff --git a/shotover/Cargo.toml b/shotover/Cargo.toml
@@ -118,7 +118,7 @@ aws-config = { version = "1.0.0", optional = true }
 aws-sdk-kms = { version = "1.1.0", optional = true }
 chacha20poly1305 = { version = "0.10.0", features = ["std"], optional = true }
 generic-array = { version = "0.14", features = ["serde"], optional = true }
-kafka-protocol = { version = "0.11.0", optional = true }
+kafka-protocol = { version = "0.12.0", optional = true, default-features = false, features = ["messages_enums", "broker", "client"] }
 rustls = { version = "0.23.0", default-features = false, features = ["tls12"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["ring"] }
 rustls-pemfile = "2.0.0"

diff --git a/shotover/src/connection.rs b/shotover/src/connection.rs
@@ -238,7 +238,7 @@ impl RequestPending {
     }
 }
 
-#[allow(clippy::too_many_arguments)]
+#[expect(clippy::too_many_arguments)]
 fn spawn_read_write_tasks<
     C: CodecBuilder + 'static,
     R: AsyncRead + Unpin + Send + 'static,

diff --git a/shotover/src/frame/kafka.rs b/shotover/src/frame/kafka.rs
diff --git a/shotover/src/frame/value/cassandra.rs b/shotover/src/frame/value/cassandra.rs
@@ -241,7 +241,6 @@ fn serialize_list(cursor: &mut Cursor<&mut Vec<u8>>, values: &[GenericValue]) {
     });
 }
 
-#[allow(clippy::mutable_key_type)]
 fn serialize_set(cursor: &mut Cursor<&mut Vec<u8>>, values: &BTreeSet<GenericValue>) {
     serialize_with_length_prefix(cursor, |cursor| {
         serialize_len(cursor, values.len());
@@ -263,7 +262,6 @@ fn serialize_stringmap(cursor: &mut Cursor<&mut Vec<u8>>, values: &BTreeMap<Stri
     });
 }
 
-#[allow(clippy::mutable_key_type)]
 fn serialize_map(cursor: &mut Cursor<&mut Vec<u8>>, values: &BTreeMap<GenericValue, GenericValue>) {
     serialize_with_length_prefix(cursor, |cursor| {
         serialize_len(cursor, values.len());

diff --git a/shotover/src/lib.rs b/shotover/src/lib.rs
@@ -25,15 +25,21 @@
 //! }
 //! ```
 
-// If we absolutely need unsafe code, it should be isolated within a separate small crate that exposes a sound safe API.
-// "sound" means that it is impossible for any interaction with the public API of the crate to violate an unsafe invariant which causes UB.
-#![deny(unsafe_code)]
-// Accidentally printing would break json log output
-#![deny(clippy::print_stdout)]
-#![deny(clippy::print_stderr)]
+#![deny(
+    unsafe_code,
+    reason = r#"
+If we absolutely need unsafe code, it should be isolated within a separate small crate that exposes a sound safe API.
+"sound" means that it is impossible for any interaction with the public API of the crate to violate an unsafe invariant which causes UB."#
+)]
+#![deny(
+    clippy::print_stdout,
+    reason = "Accidentally printing would break json log output"
+)]
+#![deny(
+    clippy::print_stderr,
+    reason = "Accidentally printing would break json log output."
+)]
 // allow some clippy lints that we disagree with
-#![allow(clippy::needless_doctest_main)]
-#![allow(clippy::box_default)]
 // Allow dead code if any of the protocol features are disabled
 #![cfg_attr(
     any(

diff --git a/shotover/src/runner.rs b/shotover/src/runner.rs
@@ -72,7 +72,7 @@ pub struct Shotover {
 }
 
 impl Shotover {
-    #[allow(clippy::new_without_default)]
+    #[expect(clippy::new_without_default)]
     pub fn new() -> Self {
         if std::env::var("RUST_LIB_BACKTRACE").is_err() {
             std::env::set_var("RUST_LIB_BACKTRACE", "0");

diff --git a/shotover/src/transforms/kafka/sink_cluster/connections.rs b/shotover/src/transforms/kafka/sink_cluster/connections.rs
@@ -48,7 +48,7 @@ impl Connections {
 
     /// If a connection already exists for the requested Destination return it.
     /// Otherwise create a new connection, cache it and return it.
-    #[allow(clippy::too_many_arguments)]
+    #[expect(clippy::too_many_arguments)]
     pub async fn get_or_open_connection(
         &mut self,
         rng: &mut SmallRng,
@@ -121,7 +121,7 @@ impl Connections {
         Ok(self.connections.get_mut(&destination).unwrap())
     }
 
-    #[allow(clippy::too_many_arguments)]
+    #[expect(clippy::too_many_arguments)]
     async fn create_and_insert_connection(
         &mut self,
         rng: &mut SmallRng,

diff --git a/shotover/src/transforms/kafka/sink_cluster/mod.rs b/shotover/src/transforms/kafka/sink_cluster/mod.rs
@@ -169,7 +169,7 @@ struct KafkaSinkClusterBuilder {
 }
 
 impl KafkaSinkClusterBuilder {
-    #[allow(clippy::too_many_arguments)]
+    #[expect(clippy::too_many_arguments)]
     pub fn new(
         chain_name: String,
         first_contact_points: Vec<KafkaAddress>,

diff --git a/shotover/src/transforms/kafka/sink_cluster/scram_over_mtls.rs b/shotover/src/transforms/kafka/sink_cluster/scram_over_mtls.rs
@@ -35,7 +35,6 @@ pub struct TokenTask {
 }
 
 impl TokenTask {
-    #[allow(clippy::new_without_default)]
     pub fn new(
         mtls_connection_factory: ConnectionFactory,
         mtls_port_contact_points: Vec<KafkaAddress>,

diff --git a/shotover/src/transforms/mod.rs b/shotover/src/transforms/mod.rs
@@ -54,7 +54,6 @@ pub struct TransformContextBuilder {
     pub client_details: String,
 }
 
-#[allow(clippy::new_without_default)]
 impl TransformContextBuilder {
     pub fn new_test() -> Self {
         TransformContextBuilder {

diff --git a/shotover/src/transforms/redis/cluster_ports_rewrite.rs b/shotover/src/transforms/redis/cluster_ports_rewrite.rs
@@ -201,7 +201,7 @@ fn rewrite_port_node(frame: &mut Frame, new_port: u16) -> Result<()> {
                     .next()
                     .ok_or_else(|| anyhow!("CLUSTER NODES response missing address field"))?;
 
-                let split = ip.split(|c| c == ':' || c == '@').collect::<Vec<&str>>();
+                let split = ip.split([':', '@']).collect::<Vec<&str>>();
 
                 if split.len() < 3 {
                     bail!("IP address not in valid format: {ip}");

diff --git a/test-helpers/src/connection/java.rs b/test-helpers/src/connection/java.rs
@@ -239,7 +239,7 @@ impl Jvm {
         Value {
             // TODO: Discuss with upstream why this was deprecated,
             // `Jvm::java_list` is very difficult to use due to Result in input.
-            #[allow(deprecated)]
+            #[expect(deprecated)]
             instance: self.0.create_java_list(element_type, &args).unwrap(),
             jvm: self.0.clone(),
         }