fix: can not connect to ZooKeeper 3.3

deserialization of `ConnectResponse.readonly` is simply wrong.

Author: kezhuw

src/proto/connect.rs

@@ -64,12 +64,12 @@ impl<'a> DeserializableRecord<'a> for ConnectResponse<'a> {
             )));
         }
         let len = unsafe { buf.get_unchecked_i32() };
-        if len <= 0 || len != (buf.len() - 1) as i32 {
+        if len <= 0 || len > buf.len() as i32 {
             return Err(DeserializeError::UnmarshalError(format!("invalid session password length {len}")));
         }
         let len = len as usize;
         let password = unsafe { buf.get_unchecked(..len) };
-        let readonly = unsafe { *buf.get_unchecked(len) };
+        let readonly = if len == buf.len() { 0 } else { unsafe { *buf.get_unchecked(len) } };
         if readonly != 0 && readonly != 1 {
             return Err(DeserializeError::UnmarshalError(format!("invalid session readonly value {readonly}")));
         }

tests/zookeeper.rs

@@ -45,8 +45,8 @@ fn zookeeper_image() -> GenericImage {
     zookeeper_image_with_properties(Vec::default())
 }
 
-fn zookeeper34_image() -> GenericImage {
-    zookeeper_image_with_version_and_properties("3.4", Vec::default())
+fn zookeeper_image_with_version<'a>(version: &'a str) -> GenericImage {
+    zookeeper_image_with_version_and_properties(version, Vec::default())
 }
 
 async fn example() {
@@ -621,10 +621,12 @@ async fn test_create_container() {
     assert_that!(client.delete("/container", None).await.unwrap_err()).is_equal_to(zk::Error::NoNode);
 }
 
+#[test_case("3.3"; "3.3")]
+#[test_case("3.4"; "3.4")]
 #[test_log::test(tokio::test)]
-async fn test_zookeeper34() {
+async fn test_zookeeper_old_server(version: &'static str) {
     let docker = DockerCli::default();
-    let zookeeper = docker.run(zookeeper34_image());
+    let zookeeper = docker.run(zookeeper_image_with_version(version));
     let zk_port = zookeeper.get_host_port(2181);
     let cluster = format!("127.0.0.1:{}", zk_port);
     let client = zk::Client::builder().assume_server_version(3, 4, u32::MAX).connect(&cluster).await.unwrap();